Merge branch 'fix/streaming-api-for-create-activity' into 'develop'

fix: stream out Create Activity

Closes #2691

See merge request pleroma/pleroma!3499

diff --combined lib/pleroma/web/activity_pub/side_effects.ex
index dda48ea5f2e1d67d4257b1ff8d8a2f96ded26249,b82a89896b4e5518cbd4f95a4a8c04b8143c7e2b..701181a1419e2f12bce3932fba3d0014832019da
--- 1/lib/pleroma/web/activity_pub/side_effects.ex
--- 2/lib/pleroma/web/activity_pub/side_effects.ex
+++ b/lib/pleroma/web/activity_pub/side_effects.ex
@@@ -10,7 -10,6 +10,6 @@@ defmodule Pleroma.Web.ActivityPub.SideE
    collection, and so on.
    """
    alias Pleroma.Activity
-   alias Pleroma.Activity.Ir.Topics
    alias Pleroma.Chat
    alias Pleroma.Chat.MessageReference
    alias Pleroma.FollowingRelationship
@@@ -24,7 -23,6 +23,7 @@@
    alias Pleroma.Web.ActivityPub.Utils
    alias Pleroma.Web.Push
    alias Pleroma.Web.Streamer
 +  alias Pleroma.Workers.PollWorker
  
    require Logger
  
@@@ -196,7 -194,7 +195,7 @@@
    # - Set up notifications
    @impl true
    def handle(%{data: %{"type" => "Create"}} = activity, meta) do
 -    with {:ok, object, meta} <- handle_object_creation(meta[:object_data], meta),
 +    with {:ok, object, meta} <- handle_object_creation(meta[:object_data], activity, meta),
           %User{} = user <- User.get_cached_by_ap_id(activity.data["actor"]) do
        {:ok, notifications} = Notification.create_notifications(activity, do_send: false)
        {:ok, _user} = ActivityPub.increase_note_count_if_public(user, object)
@@@ -226,6 -224,8 +225,8 @@@
          meta
          |> add_notifications(notifications)
  
+       ap_streamer().stream_out(activity)
+ 
        {:ok, activity, meta}
      else
        e -> Repo.rollback(e)
@@@ -246,9 -246,7 +247,7 @@@
      if !User.is_internal_user?(user) do
        Notification.create_notifications(object)
  
-       object
-       |> Topics.get_activity_topics()
-       |> Streamer.stream(object)
+       ap_streamer().stream_out(object)
      end
  
      {:ok, object, meta}
@@@ -390,7 -388,7 +389,7 @@@
      {:ok, object, meta}
    end
  
 -  def handle_object_creation(%{"type" => "ChatMessage"} = object, meta) do
 +  def handle_object_creation(%{"type" => "ChatMessage"} = object, _activity, meta) do
      with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
        actor = User.get_cached_by_ap_id(object.data["actor"])
        recipient = User.get_cached_by_ap_id(hd(object.data["to"]))
@@@ -425,14 -423,7 +424,14 @@@
      end
    end
  
 -  def handle_object_creation(%{"type" => "Answer"} = object_map, meta) do
 +  def handle_object_creation(%{"type" => "Question"} = object, activity, meta) do
 +    with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
 +      PollWorker.schedule_poll_end(activity)
 +      {:ok, object, meta}
 +    end
 +  end
 +
 +  def handle_object_creation(%{"type" => "Answer"} = object_map, _activity, meta) do
      with {:ok, object, meta} <- Pipeline.common_pipeline(object_map, meta) do
        Object.increase_vote_count(
          object.data["inReplyTo"],
@@@ -444,15 -435,15 +443,15 @@@
      end
    end
  
 -  def handle_object_creation(%{"type" => objtype} = object, meta)
 -      when objtype in ~w[Audio Video Question Event Article Note Page] do
 +  def handle_object_creation(%{"type" => objtype} = object, _activity, meta)
 +      when objtype in ~w[Audio Video Event Article Note Page] do
      with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
        {:ok, object, meta}
      end
    end
  
    # Nothing to do
 -  def handle_object_creation(object, meta) do
 +  def handle_object_creation(object, _activity, meta) do
      {:ok, object, meta}
    end