X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Fstreamer.ex;h=6b6d40346191c00c662077c57894ff34a14852ae;hb=b4bd5e40e491c8b777c75ec4f096c0c466a6b4e2;hp=6aac472dcdefbc6486d0d0151a477b1eb1b146b7;hpb=4fd9df100f61dfe4731934eca86500472d866f60;p=akkoma diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex index 6aac472dc..6b6d40346 100644 --- a/lib/pleroma/web/streamer.ex +++ b/lib/pleroma/web/streamer.ex @@ -1,7 +1,8 @@ defmodule Pleroma.Web.Streamer do use GenServer require Logger - alias Pleroma.{User, Notification} + alias Pleroma.{User, Notification, Activity, Object, Repo} + alias Pleroma.Web.ActivityPub.ActivityPub def init(args) do {:ok, args} @@ -59,6 +60,35 @@ defmodule Pleroma.Web.Streamer do {:noreply, topics} end + def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do + author = User.get_cached_by_ap_id(item.data["actor"]) + + # filter the recipient list if the activity is not public, see #270. + recipient_lists = + case ActivityPub.is_public?(item) do + true -> + Pleroma.List.get_lists_from_activity(item) + + _ -> + Pleroma.List.get_lists_from_activity(item) + |> Enum.filter(fn list -> + owner = Repo.get(User, list.user_id) + author.follower_address in owner.following + 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(topics, list_topic, item) + end) + + {:noreply, topics} + end + def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do topic = "user:#{item.user_id}" @@ -125,6 +155,34 @@ defmodule Pleroma.Web.Streamer do {:noreply, state} end + defp represent_update(%Activity{} = activity, %User{} = user) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity, + for: user + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do + Enum.each(topics[topic] || [], fn socket -> + # Get the current user so we have up-to-date blocks etc. + user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id) + blocks = user.info["blocks"] || [] + + parent = Object.normalize(item.data["object"]) + + unless is_nil(parent) or item.actor in blocks or parent.data["actor"] in blocks do + send(socket.transport_pid, {:text, represent_update(item, user)}) + end + end) + end + def push_to_socket(topics, topic, item) do Enum.each(topics[topic] || [], fn socket -> # Get the current user so we have up-to-date blocks etc. @@ -132,25 +190,12 @@ defmodule Pleroma.Web.Streamer do blocks = user.info["blocks"] || [] unless item.actor in blocks do - json = - %{ - event: "update", - payload: - Pleroma.Web.MastodonAPI.StatusView.render( - "status.json", - activity: item, - for: user - ) - |> Jason.encode!() - } - |> Jason.encode!() - - send(socket.transport_pid, {:text, json}) + send(socket.transport_pid, {:text, represent_update(item, user)}) end end) end - defp internal_topic(topic, socket) when topic in ~w[user, direct] do + defp internal_topic(topic, socket) when topic in ~w[user direct] do "#{topic}:#{socket.assigns[:user].id}" end