Stream out edits

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -194,7 +194,16 @@ defp insert_activity_with_expiration(data, local, recipients) do
   def notify_and_stream(activity) do
     Notification.create_notifications(activity)
 
-    conversation = create_or_bump_conversation(activity, activity.actor)
+    original_activity =
+      case activity do
+        %{data: %{"type" => "Update"}, object: %{data: %{"id" => id}}} ->
+          Activity.get_create_by_object_ap_id_with_object(id)
+
+        _ ->
+          activity
+      end
+
+    conversation = create_or_bump_conversation(original_activity, original_activity.actor)
     participations = get_participations(conversation)
     stream_out(activity)
     stream_out_participations(participations)
@@ -260,7 +269,7 @@ def stream_out_participations(_, _), do: :noop
 
   @impl true
   def stream_out(%Activity{data: %{"type" => data_type}} = activity)
-      when data_type in ["Create", "Announce", "Delete"] do
+      when data_type in ["Create", "Announce", "Delete", "Update"] do
     activity
     |> Topics.get_activity_topics()
     |> Streamer.stream(activity)

lib/pleroma/web/activity_pub/side_effects.ex

@@ -477,6 +477,12 @@ defp handle_update_object(
       |> Repo.preload(:hashtags)
       |> Object.change(%{data: updated_object_data})
       |> Object.update_and_set_cache()
+
+      if updated do
+        object
+        |> Activity.normalize()
+        |> ActivityPub.notify_and_stream()
+      end
     end
 
     {:ok, object, meta}

lib/pleroma/web/streamer.ex

@@ -287,6 +287,20 @@ defp push_to_socket(topic, %Activity{
 
   defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
 
+  defp push_to_socket(topic, %Activity{data: %{"type" => "Update"}} = item) do
+    anon_render = StreamerView.render("status_update.json", item)
+
+    Registry.dispatch(@registry, topic, fn list ->
+      Enum.each(list, fn {pid, auth?} ->
+        if auth? do
+          send(pid, {:render_with_user, StreamerView, "status_update.json", item})
+        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)
 

lib/pleroma/web/views/streamer_view.ex

@@ -26,6 +26,23 @@ def render("update.json", %Activity{} = activity, %User{} = user, topic) do
     |> Jason.encode!()
   end
 
+  def render("status_update.json", %Activity{} = activity, %User{} = user, topic) do
+    activity = Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
+
+    %{
+      stream: [topic],
+      event: "status.update",
+      payload:
+        Pleroma.Web.MastodonAPI.StatusView.render(
+          "show.json",
+          activity: activity,
+          for: user
+        )
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
   def render("notification.json", %Notification{} = notify, %User{} = user, topic) do
     %{
       stream: [topic],
@@ -54,6 +71,22 @@ def render("update.json", %Activity{} = activity, topic) do
     |> Jason.encode!()
   end
 
+  def render("status_update.json", %Activity{} = activity, topic) do
+    activity = Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
+
+    %{
+      stream: [topic],
+      event: "status.update",
+      payload:
+        Pleroma.Web.MastodonAPI.StatusView.render(
+          "show.json",
+          activity: activity
+        )
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
   def render("follow_relationships_update.json", item, topic) do
     %{
       stream: [topic],

test/pleroma/web/streamer_test.exs

@@ -383,6 +383,20 @@ test "it sends follow relationships updates to the 'user' stream", %{
                "state" => "follow_accept"
              } = Jason.decode!(payload)
     end
+
+    test "it streams edits in the 'user' stream", %{user: user, token: oauth_token} do
+      sender = insert(:user)
+      {:ok, _, _, _} = CommonAPI.follow(user, sender)
+
+      {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
+
+      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})
+      edited = Pleroma.Activity.normalize(edited)
+
+      assert_receive {:render_with_user, _, "status_update.json", ^edited}
+      refute Streamer.filtered_by_user?(user, edited)
+    end
   end
 
   describe "public streams" do
@@ -425,6 +439,25 @@ test "handles deletions" do
       assert_receive {:text, event}
       assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
     end
+
+    test "it streams edits in the 'public' stream" do
+      sender = insert(:user)
+
+      Streamer.get_topic_and_add_socket("public", nil, nil)
+      {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
+      assert_receive {:text, _}
+
+      {:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})
+
+      edited = Pleroma.Activity.normalize(edited)
+
+      %{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])
+
+      assert_receive {:text, event}
+      assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
+      assert %{"id" => ^activity_id} = Jason.decode!(payload)
+      refute Streamer.filtered_by_user?(sender, edited)
+    end
   end
 
   describe "thread_containment/2" do