use GenServer
require Logger
alias Pleroma.Activity
+ alias Pleroma.Config
+ alias Pleroma.Conversation.Participation
alias Pleroma.Notification
alias Pleroma.Object
- alias Pleroma.Repo
alias Pleroma.User
alias Pleroma.Web.ActivityPub.ActivityPub
alias Pleroma.Web.ActivityPub.Visibility
+ alias Pleroma.Web.CommonAPI
alias Pleroma.Web.MastodonAPI.NotificationView
@keepalive_interval :timer.seconds(30)
{:noreply, topics}
end
+ def handle_cast(%{action: :stream, topic: "participation", item: participation}, topics) do
+ user_topic = "direct:#{participation.user_id}"
+ Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
+
+ push_to_socket(topics, user_topic, participation)
+
+ {:noreply, topics}
+ end
+
def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
# filter the recipient list if the activity is not public, see #270.
recipient_lists =
_ ->
Pleroma.List.get_lists_from_activity(item)
|> Enum.filter(fn list ->
- owner = Repo.get(User, list.user_id)
+ owner = User.get_cached_by_id(list.user_id)
Visibility.visible_for_user?(item, owner)
end)
{:noreply, topics}
end
- def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do
- topic = "user:#{item.user_id}"
-
- Enum.each(topics[topic] || [], fn socket ->
- json =
- %{
- event: "notification",
- payload:
- NotificationView.render("show.json", %{
- notification: item,
- for: socket.assigns["user"]
- })
- |> Jason.encode!()
- }
- |> Jason.encode!()
-
- send(socket.transport_pid, {:text, json})
+ def handle_cast(
+ %{action: :stream, topic: topic, item: %Notification{} = item},
+ topics
+ )
+ when topic in ["user", "user:notification"] do
+ topics
+ |> Map.get("#{topic}:#{item.user_id}", [])
+ |> Enum.each(fn socket ->
+ with %User{} = user <- User.get_cached_by_ap_id(socket.assigns[:user].ap_id),
+ true <- should_send?(user, item),
+ false <- CommonAPI.thread_muted?(user, item.activity) do
+ send(
+ socket.transport_pid,
+ {:text, represent_notification(socket.assigns[:user], item)}
+ )
+ end
end)
{:noreply, topics}
|> Jason.encode!()
end
+ def represent_conversation(%Participation{} = participation) do
+ %{
+ event: "conversation",
+ payload:
+ Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
+ participation: participation,
+ for: participation.user
+ })
+ |> Jason.encode!()
+ }
+ |> Jason.encode!()
+ end
+
+ @spec represent_notification(User.t(), Notification.t()) :: binary()
+ defp represent_notification(%User{} = user, %Notification{} = notify) do
+ %{
+ event: "notification",
+ payload:
+ NotificationView.render(
+ "show.json",
+ %{notification: notify, for: user}
+ )
+ |> Jason.encode!()
+ }
+ |> Jason.encode!()
+ end
+
+ defp should_send?(%User{} = user, %Activity{} = item) do
+ blocks = user.info.blocks || []
+ mutes = user.info.mutes || []
+ reblog_mutes = user.info.muted_reblogs || []
+ domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)
+
+ with parent when not is_nil(parent) <- Object.normalize(item),
+ true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
+ true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
+ %{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) do
+ true
+ else
+ _ -> false
+ end
+ end
+
+ defp should_send?(%User{} = user, %Notification{activity: activity}) do
+ should_send?(user, activity)
+ 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.
if socket.assigns[:user] do
user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
- blocks = user.info.blocks || []
- mutes = user.info.mutes || []
- reblog_mutes = user.info.muted_reblogs || []
-
- parent = Object.normalize(item)
- unless is_nil(parent) or item.actor in blocks or item.actor in mutes or
- item.actor in reblog_mutes or not ActivityPub.contain_activity(item, user) or
- parent.data["actor"] in blocks or parent.data["actor"] in mutes do
+ if should_send?(user, item) do
send(socket.transport_pid, {:text, represent_update(item, user)})
end
else
end)
end
+ def push_to_socket(topics, topic, %Participation{} = participation) do
+ Enum.each(topics[topic] || [], fn socket ->
+ send(socket.transport_pid, {:text, represent_conversation(participation)})
+ end)
+ end
+
def push_to_socket(topics, topic, %Activity{
data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
}) do
blocks = user.info.blocks || []
mutes = user.info.mutes || []
- unless item.actor in blocks or item.actor in mutes or
- not ActivityPub.contain_activity(item, user) do
+ with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
+ true <- thread_containment(item, user) do
send(socket.transport_pid, {:text, represent_update(item, user)})
end
else
end)
end
- defp internal_topic(topic, socket) when topic in ~w[user direct] do
+ defp internal_topic(topic, socket) when topic in ~w[user user:notification direct] do
"#{topic}:#{socket.assigns[:user].id}"
end
defp internal_topic(topic, _), do: topic
+
+ @spec thread_containment(Activity.t(), User.t()) :: boolean()
+ defp thread_containment(_activity, %User{info: %{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
end
projects / akkoma / blobdiff