# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only

defmodule Pleroma.Web.Streamer do
  require Logger
  require Pleroma.Constants

  alias Pleroma.Activity
  alias Pleroma.Chat.MessageReference
  alias Pleroma.Config
  alias Pleroma.Conversation.Participation
  alias Pleroma.Notification
  alias Pleroma.Object
  alias Pleroma.User
  alias Pleroma.Web.ActivityPub.ActivityPub
  alias Pleroma.Web.ActivityPub.Visibility
  alias Pleroma.Web.CommonAPI
  alias Pleroma.Web.OAuth.Token
  alias Pleroma.Web.Plugs.OAuthScopesPlug
  alias Pleroma.Web.StreamerView

  @registry Pleroma.Web.StreamerRegistry

  def registry, do: @registry

  @public_streams Pleroma.Constants.public_streams()
  @local_streams ["public:local", "public:local:media"]
  @user_streams ["user", "user:notification", "direct", "user:pleroma_chat"]

  @doc "Expands and authorizes a stream, and registers the process for streaming."
  @spec get_topic_and_add_socket(
          stream :: String.t(),
          User.t() | nil,
          Token.t() | nil,
          map() | nil
        ) ::
          {:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
  def get_topic_and_add_socket(stream, user, oauth_token, params \\ %{}) do
    with {:ok, topic} <- get_topic(stream, user, oauth_token, params) do
      add_socket(topic, oauth_token)
    end
  end

  defp can_access_stream(user, oauth_token, kind) do
    with {_, true} <- {:restrict?, Config.restrict_unauthenticated_access?(:timelines, kind)},
         {_, %User{id: user_id}, %Token{user_id: user_id}} <- {:user, user, oauth_token},
         {_, true} <-
           {:scopes,
            OAuthScopesPlug.filter_descendants(["read:statuses"], oauth_token.scopes) != []} do
      true
    else
      {:restrict?, _} ->
        true

      _ ->
        false
    end
  end

  @doc "Expand and authorizes a stream"
  @spec get_topic(stream :: String.t() | nil, User.t() | nil, Token.t() | nil, map()) ::
          {:ok, topic :: String.t() | nil} | {:error, :bad_topic}
  def get_topic(stream, user, oauth_token, params \\ %{})

  def get_topic(nil = _stream, _user, _oauth_token, _params) do
    {:ok, nil}
  end

  # Allow all public steams if the instance allows unauthenticated access.
  # Otherwise, only allow users with valid oauth tokens.
  def get_topic(stream, user, oauth_token, _params) when stream in @public_streams do
    kind = if stream in @local_streams, do: :local, else: :federated

    if can_access_stream(user, oauth_token, kind) do
      {:ok, stream}
    else
      {:error, :unauthorized}
    end
  end

  # Allow all hashtags streams.
  def get_topic("hashtag", _user, _oauth_token, %{"tag" => tag} = _params) do
    {:ok, "hashtag:" <> tag}
  end

  # Allow remote instance streams.
  def get_topic("public:remote", user, oauth_token, %{"instance" => instance} = _params) do
    if can_access_stream(user, oauth_token, :federated) do
      {:ok, "public:remote:" <> instance}
    else
      {:error, :unauthorized}
    end
  end

  def get_topic("public:remote:media", user, oauth_token, %{"instance" => instance} = _params) do
    if can_access_stream(user, oauth_token, :federated) do
      {:ok, "public:remote:media:" <> instance}
    else
      {:error, :unauthorized}
    end
  end

  # Expand user streams.
  def get_topic(
        stream,
        %User{id: user_id} = user,
        %Token{user_id: user_id} = oauth_token,
        _params
      )
      when stream in @user_streams do
    # Note: "read" works for all user streams (not mentioning it since it's an ancestor scope)
    required_scopes =
      if stream == "user:notification" do
        ["read:notifications"]
      else
        ["read:statuses"]
      end

    if OAuthScopesPlug.filter_descendants(required_scopes, oauth_token.scopes) == [] do
      {:error, :unauthorized}
    else
      {:ok, stream <> ":" <> to_string(user.id)}
    end
  end

  def get_topic(stream, _user, _oauth_token, _params) when stream in @user_streams do
    {:error, :unauthorized}
  end

  # List streams.
  def get_topic(
        "list",
        %User{id: user_id} = user,
        %Token{user_id: user_id} = oauth_token,
        %{"list" => id}
      ) do
    cond do
      OAuthScopesPlug.filter_descendants(["read", "read:lists"], oauth_token.scopes) == [] ->
        {:error, :unauthorized}

      Pleroma.List.get(id, user) ->
        {:ok, "list:" <> to_string(id)}

      true ->
        {:error, :bad_topic}
    end
  end

  def get_topic("list", _user, _oauth_token, _params) do
    {:error, :unauthorized}
  end

  def get_topic(_stream, _user, _oauth_token, _params) do
    {:error, :bad_topic}
  end

  @doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
  def add_socket(topic, oauth_token) do
    if should_env_send?() do
      oauth_token_id = if oauth_token, do: oauth_token.id, else: false
      Registry.register(@registry, topic, oauth_token_id)
    end

    {:ok, topic}
  end

  def remove_socket(topic) do
    if should_env_send?(), do: Registry.unregister(@registry, topic)
  end

  def stream(topics, items) do
    if should_env_send?() do
      for topic <- List.wrap(topics), item <- List.wrap(items) do
        spawn(fn -> do_stream(topic, item) end)
      end
    end
  end

  def filtered_by_user?(user, item, streamed_type \\ :activity)

  def filtered_by_user?(%User{} = user, %Activity{} = item, streamed_type) do
    %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
      User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])

    recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
    recipients = MapSet.new(item.recipients)
    domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)

    with parent <- Object.normalize(item, fetch: false) || item,
         true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
         true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
         true <-
           !(streamed_type == :activity && item.data["type"] == "Announce" &&
               parent.data["actor"] == user.ap_id),
         true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
         true <- MapSet.disjoint?(recipients, recipient_blocks),
         %{host: item_host} <- URI.parse(item.actor),
         %{host: parent_host} <- URI.parse(parent.data["actor"]),
         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
         true <- thread_containment(item, user),
         false <- CommonAPI.thread_muted?(user, parent) do
      false
    else
      _ -> true
    end
  end

  def filtered_by_user?(%User{} = user, %Notification{activity: activity}, _) do
    filtered_by_user?(user, activity, :notification)
  end

  defp do_stream("direct", item) do
    recipient_topics =
      User.get_recipients_from_activity(item)
      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)

    Enum.each(recipient_topics, fn user_topic ->
      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
      push_to_socket(user_topic, item)
    end)
  end

  defp do_stream("follow_relationship", item) do
    user_topic = "user:#{item.follower.id}"
    text = StreamerView.render("follow_relationships_update.json", item, user_topic)

    Logger.debug("Trying to push follow relationship update to #{user_topic}\n\n")

    Registry.dispatch(@registry, user_topic, fn list ->
      Enum.each(list, fn {pid, _auth} ->
        send(pid, {:text, text})
      end)
    end)
  end

  defp do_stream("participation", participation) do
    user_topic = "direct:#{participation.user_id}"
    Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

    push_to_socket(user_topic, participation)
  end

  defp do_stream("list", item) do
    # filter the recipient list if the activity is not public, see #270.
    recipient_lists =
      case Visibility.is_public?(item) do
        true ->
          Pleroma.List.get_lists_from_activity(item)

        _ ->
          Pleroma.List.get_lists_from_activity(item)
          |> Enum.filter(fn list ->
            owner = User.get_cached_by_id(list.user_id)

            Visibility.visible_for_user?(item, owner)
          end)
      end

    recipient_topics =
      recipient_lists
      |> Enum.map(fn %{id: id} -> "list:#{id}" end)

    Enum.each(recipient_topics, fn list_topic ->
      Logger.debug("Trying to push message to #{list_topic}\n\n")
      push_to_socket(list_topic, item)
    end)
  end

  defp do_stream(topic, %Notification{} = item)
       when topic in ["user", "user:notification"] do
    user_topic = "#{topic}:#{item.user_id}"

    Registry.dispatch(@registry, user_topic, fn list ->
      Enum.each(list, fn {pid, _auth} ->
        send(pid, {:render_with_user, StreamerView, "notification.json", item, user_topic})
      end)
    end)
  end

  defp do_stream(topic, {user, %MessageReference{} = cm_ref})
       when topic in ["user", "user:pleroma_chat"] do
    topic = "#{topic}:#{user.id}"

    text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref}, topic)

    Registry.dispatch(@registry, topic, fn list ->
      Enum.each(list, fn {pid, _auth} ->
        send(pid, {:text, text})
      end)
    end)
  end

  defp do_stream("user", item) do
    Logger.debug("Trying to push to users")

    recipient_topics =
      User.get_recipients_from_activity(item)
      |> Enum.map(fn %{id: id} -> "user:#{id}" end)

    Enum.each(recipient_topics, fn topic ->
      push_to_socket(topic, item)
    end)
  end

  defp do_stream(topic, item) do
    Logger.debug("Trying to push to #{topic}")
    Logger.debug("Pushing item to #{topic}")
    push_to_socket(topic, item)
  end

  defp push_to_socket(topic, %Participation{} = participation) do
    rendered = StreamerView.render("conversation.json", participation, topic)

    Registry.dispatch(@registry, topic, fn list ->
      Enum.each(list, fn {pid, _} ->
        send(pid, {:text, rendered})
      end)
    end)
  end

  defp push_to_socket(topic, %Activity{
         data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
       }) do
    rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})

    Registry.dispatch(@registry, topic, fn list ->
      Enum.each(list, fn {pid, _} ->
        send(pid, {:text, rendered})
      end)
    end)
  end

  defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop

  defp push_to_socket(topic, %Activity{data: %{"type" => "Update"}} = item) do
    create_activity =
      Pleroma.Activity.get_create_by_object_ap_id(item.object.data["id"])
      |> Map.put(:object, item.object)

    anon_render = StreamerView.render("status_update.json", create_activity, topic)

    Registry.dispatch(@registry, topic, fn list ->
      Enum.each(list, fn {pid, auth?} ->
        if auth? do
          send(
            pid,
            {:render_with_user, StreamerView, "status_update.json", create_activity, topic}
          )
        else
          send(pid, {:text, anon_render})
        end
      end)
    end)
  end

  defp push_to_socket(topic, item) do
    anon_render = StreamerView.render("update.json", item, topic)

    Registry.dispatch(@registry, topic, fn list ->
      Enum.each(list, fn {pid, auth?} ->
        if auth? do
          send(pid, {:render_with_user, StreamerView, "update.json", item, topic})
        else
          send(pid, {:text, anon_render})
        end
      end)
    end)
  end

  defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true

  defp thread_containment(activity, user) do
    if Config.get([:instance, :skip_thread_containment]) do
      true
    else
      ActivityPub.contain_activity(activity, user)
    end
  end

  def close_streams_by_oauth_token(oauth_token) do
    if should_env_send?() do
      Registry.select(
        @registry,
        [
          {
            {:"$1", :"$2", :"$3"},
            [{:==, :"$3", oauth_token.id}],
            [:"$2"]
          }
        ]
      )
      |> Enum.each(fn pid -> send(pid, :close) end)
    end
  end

  # In dev/prod the streamer registry is expected to be started, so return true
  # In test it is possible to have the registry started for a test so it will check
  # In benchmark it will never find the process alive and return false
  def should_env_send? do
    if Application.get_env(:pleroma, Pleroma.Application)[:streamer_registry] do
      true
    else
      case Process.whereis(@registry) do
        nil ->
          false

        pid ->
          Process.alive?(pid)
      end
    end
  end
end