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

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.Streamer do
   6   require Logger
   7
   8   alias Pleroma.Activity
   9   alias Pleroma.Chat.MessageReference
  10   alias Pleroma.Config
  11   alias Pleroma.Conversation.Participation
  12   alias Pleroma.Notification
  13   alias Pleroma.Object
  14   alias Pleroma.User
  15   alias Pleroma.Web.ActivityPub.ActivityPub
  16   alias Pleroma.Web.ActivityPub.Visibility
  17   alias Pleroma.Web.CommonAPI
  18   alias Pleroma.Web.StreamerView
  19
  20   @mix_env Mix.env()
  21   @registry Pleroma.Web.StreamerRegistry
  22
  23   def registry, do: @registry
  24
  25   @public_streams ["public", "public:local", "public:media", "public:local:media"]
  26   @user_streams ["user", "user:notification", "direct", "user:pleroma_chat"]
  27
  28   @doc "Expands and authorizes a stream, and registers the process for streaming."
  29   @spec get_topic_and_add_socket(stream :: String.t(), User.t() | nil, Map.t() | nil) ::
  30           {:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
  31   def get_topic_and_add_socket(stream, user, params \\ %{}) do
  32     case get_topic(stream, user, params) do
  33       {:ok, topic} -> add_socket(topic, user)
  34       error -> error
  35     end
  36   end
  37
  38   @doc "Expand and authorizes a stream"
  39   @spec get_topic(stream :: String.t(), User.t() | nil, Map.t()) ::
  40           {:ok, topic :: String.t()} | {:error, :bad_topic}
  41   def get_topic(stream, user, params \\ %{})
  42
  43   # Allow all public steams.
  44   def get_topic(stream, _, _) when stream in @public_streams do
  45     {:ok, stream}
  46   end
  47
  48   # Allow all hashtags streams.
  49   def get_topic("hashtag", _, %{"tag" => tag}) do
  50     {:ok, "hashtag:" <> tag}
  51   end
  52
  53   # Expand user streams.
  54   def get_topic(stream, %User{} = user, _) when stream in @user_streams do
  55     {:ok, stream <> ":" <> to_string(user.id)}
  56   end
  57
  58   def get_topic(stream, _, _) when stream in @user_streams do
  59     {:error, :unauthorized}
  60   end
  61
  62   # List streams.
  63   def get_topic("list", %User{} = user, %{"list" => id}) do
  64     if Pleroma.List.get(id, user) do
  65       {:ok, "list:" <> to_string(id)}
  66     else
  67       {:error, :bad_topic}
  68     end
  69   end
  70
  71   def get_topic("list", _, _) do
  72     {:error, :unauthorized}
  73   end
  74
  75   def get_topic(_, _, _) do
  76     {:error, :bad_topic}
  77   end
  78
  79   @doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
  80   def add_socket(topic, user) do
  81     if should_env_send?() do
  82       auth? = if user, do: true
  83       Registry.register(@registry, topic, auth?)
  84     end
  85
  86     {:ok, topic}
  87   end
  88
  89   def remove_socket(topic) do
  90     if should_env_send?(), do: Registry.unregister(@registry, topic)
  91   end
  92
  93   def stream(topics, items) do
  94     if should_env_send?() do
  95       List.wrap(topics)
  96       |> Enum.each(fn topic ->
  97         List.wrap(items)
  98         |> Enum.each(fn item ->
  99           spawn(fn -> do_stream(topic, item) end)
 100         end)
 101       end)
 102     end
 103
 104     :ok
 105   end
 106
 107   def filtered_by_user?(user, item, streamed_type \\ :activity)
 108
 109   def filtered_by_user?(%User{} = user, %Activity{} = item, streamed_type) do
 110     %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
 111       User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])
 112
 113     recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
 114     recipients = MapSet.new(item.recipients)
 115     domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
 116
 117     with parent <- Object.normalize(item) || item,
 118          true <-
 119            Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
 120          true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
 121          true <-
 122            !(streamed_type == :activity && item.data["type"] == "Announce" &&
 123                parent.data["actor"] == user.ap_id),
 124          true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
 125          true <- MapSet.disjoint?(recipients, recipient_blocks),
 126          %{host: item_host} <- URI.parse(item.actor),
 127          %{host: parent_host} <- URI.parse(parent.data["actor"]),
 128          false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
 129          false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
 130          true <- thread_containment(item, user),
 131          false <- CommonAPI.thread_muted?(user, parent) do
 132       false
 133     else
 134       _ -> true
 135     end
 136   end
 137
 138   def filtered_by_user?(%User{} = user, %Notification{activity: activity}, _) do
 139     filtered_by_user?(user, activity, :notification)
 140   end
 141
 142   defp do_stream("direct", item) do
 143     recipient_topics =
 144       User.get_recipients_from_activity(item)
 145       |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
 146
 147     Enum.each(recipient_topics, fn user_topic ->
 148       Logger.debug("Trying to push direct message to #{user_topic}\n\n")
 149       push_to_socket(user_topic, item)
 150     end)
 151   end
 152
 153   defp do_stream("participation", participation) do
 154     user_topic = "direct:#{participation.user_id}"
 155     Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
 156
 157     push_to_socket(user_topic, participation)
 158   end
 159
 160   defp do_stream("list", item) do
 161     # filter the recipient list if the activity is not public, see #270.
 162     recipient_lists =
 163       case Visibility.is_public?(item) do
 164         true ->
 165           Pleroma.List.get_lists_from_activity(item)
 166
 167         _ ->
 168           Pleroma.List.get_lists_from_activity(item)
 169           |> Enum.filter(fn list ->
 170             owner = User.get_cached_by_id(list.user_id)
 171
 172             Visibility.visible_for_user?(item, owner)
 173           end)
 174       end
 175
 176     recipient_topics =
 177       recipient_lists
 178       |> Enum.map(fn %{id: id} -> "list:#{id}" end)
 179
 180     Enum.each(recipient_topics, fn list_topic ->
 181       Logger.debug("Trying to push message to #{list_topic}\n\n")
 182       push_to_socket(list_topic, item)
 183     end)
 184   end
 185
 186   defp do_stream(topic, %Notification{} = item)
 187        when topic in ["user", "user:notification"] do
 188     Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
 189       Enum.each(list, fn {pid, _auth} ->
 190         send(pid, {:render_with_user, StreamerView, "notification.json", item})
 191       end)
 192     end)
 193   end
 194
 195   defp do_stream(topic, {user, %MessageReference{} = cm_ref})
 196        when topic in ["user", "user:pleroma_chat"] do
 197     topic = "#{topic}:#{user.id}"
 198
 199     text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
 200
 201     Registry.dispatch(@registry, topic, fn list ->
 202       Enum.each(list, fn {pid, _auth} ->
 203         send(pid, {:text, text})
 204       end)
 205     end)
 206   end
 207
 208   defp do_stream("user", item) do
 209     Logger.debug("Trying to push to users")
 210
 211     recipient_topics =
 212       User.get_recipients_from_activity(item)
 213       |> Enum.map(fn %{id: id} -> "user:#{id}" end)
 214
 215     Enum.each(recipient_topics, fn topic ->
 216       push_to_socket(topic, item)
 217     end)
 218   end
 219
 220   defp do_stream(topic, item) do
 221     Logger.debug("Trying to push to #{topic}")
 222     Logger.debug("Pushing item to #{topic}")
 223     push_to_socket(topic, item)
 224   end
 225
 226   defp push_to_socket(topic, %Participation{} = participation) do
 227     rendered = StreamerView.render("conversation.json", participation)
 228
 229     Registry.dispatch(@registry, topic, fn list ->
 230       Enum.each(list, fn {pid, _} ->
 231         send(pid, {:text, rendered})
 232       end)
 233     end)
 234   end
 235
 236   defp push_to_socket(topic, %Activity{
 237          data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
 238        }) do
 239     rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})
 240
 241     Registry.dispatch(@registry, topic, fn list ->
 242       Enum.each(list, fn {pid, _} ->
 243         send(pid, {:text, rendered})
 244       end)
 245     end)
 246   end
 247
 248   defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
 249
 250   defp push_to_socket(topic, item) do
 251     anon_render = StreamerView.render("update.json", item)
 252
 253     Registry.dispatch(@registry, topic, fn list ->
 254       Enum.each(list, fn {pid, auth?} ->
 255         if auth? do
 256           send(pid, {:render_with_user, StreamerView, "update.json", item})
 257         else
 258           send(pid, {:text, anon_render})
 259         end
 260       end)
 261     end)
 262   end
 263
 264   defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true
 265
 266   defp thread_containment(activity, user) do
 267     if Config.get([:instance, :skip_thread_containment]) do
 268       true
 269     else
 270       ActivityPub.contain_activity(activity, user)
 271     end
 272   end
 273
 274   # In test environement, only return true if the registry is started.
 275   # In benchmark environment, returns false.
 276   # In any other environment, always returns true.
 277   cond do
 278     @mix_env == :test ->
 279       def should_env_send? do
 280         case Process.whereis(@registry) do
 281           nil ->
 282             false
 283
 284           pid ->
 285             Process.alive?(pid)
 286         end
 287       end
 288
 289     @mix_env == :benchmark ->
 290       def should_env_send?, do: false
 291
 292     true ->
 293       def should_env_send?, do: true
 294   end
 295 end