Stream out edits

This commit is contained in:
Tusooa Zhu 2022-06-03 21:15:17 -04:00
parent fa31ae50e6
commit 8bac8147d4
No known key found for this signature in database
GPG Key ID: 7B467EDE43A08224
5 changed files with 95 additions and 2 deletions

View File

@ -190,7 +190,16 @@ defp insert_activity_with_expiration(data, local, recipients) do
def notify_and_stream(activity) do def notify_and_stream(activity) do
Notification.create_notifications(activity) 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) participations = get_participations(conversation)
stream_out(activity) stream_out(activity)
stream_out_participations(participations) stream_out_participations(participations)
@ -256,7 +265,7 @@ def stream_out_participations(_, _), do: :noop
@impl true @impl true
def stream_out(%Activity{data: %{"type" => data_type}} = activity) 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 activity
|> Topics.get_activity_topics() |> Topics.get_activity_topics()
|> Streamer.stream(activity) |> Streamer.stream(activity)

View File

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

View File

@ -296,6 +296,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" => "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 defp push_to_socket(topic, item) do
anon_render = StreamerView.render("update.json", item) anon_render = StreamerView.render("update.json", item)

View File

@ -25,6 +25,22 @@ def render("update.json", %Activity{} = activity, %User{} = user) do
|> Jason.encode!() |> Jason.encode!()
end end
def render("status_update.json", %Activity{} = activity, %User{} = user) do
activity = Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
%{
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) do def render("notification.json", %Notification{} = notify, %User{} = user) do
%{ %{
event: "notification", event: "notification",
@ -51,6 +67,21 @@ def render("update.json", %Activity{} = activity) do
|> Jason.encode!() |> Jason.encode!()
end end
def render("status_update.json", %Activity{} = activity) do
activity = Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
%{
event: "status.update",
payload:
Pleroma.Web.MastodonAPI.StatusView.render(
"show.json",
activity: activity
)
|> Jason.encode!()
}
|> Jason.encode!()
end
def render("chat_update.json", %{chat_message_reference: cm_ref}) do def render("chat_update.json", %{chat_message_reference: cm_ref}) do
# Explicitly giving the cmr for the object here, so we don't accidentally # Explicitly giving the cmr for the object here, so we don't accidentally
# send a later 'last_message' that was inserted between inserting this and # send a later 'last_message' that was inserted between inserting this and

View File

@ -442,6 +442,20 @@ test "it sends follow relationships updates to the 'user' stream", %{
"state" => "follow_accept" "state" => "follow_accept"
} = Jason.decode!(payload) } = Jason.decode!(payload)
end 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 end
describe "public streams" do describe "public streams" do
@ -484,6 +498,25 @@ test "handles deletions" do
assert_receive {:text, event} assert_receive {:text, event}
assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event) assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
end 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 end
describe "thread_containment/2" do describe "thread_containment/2" do