defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
+ alias Pleroma.Config
alias Pleroma.Conversation
alias Pleroma.Notification
alias Pleroma.Object
end
defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
- limit = Pleroma.Config.get([:instance, :remote_limit])
+ limit = Config.get([:instance, :remote_limit])
String.length(content) <= limit
end
end)
end
+ def stream_out_participations(%Object{data: %{"context" => context}}, user) do
+ with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
+ conversation = Repo.preload(conversation, :participations),
+ last_activity_id =
+ fetch_latest_activity_id_for_context(conversation.ap_id, %{
+ "user" => user,
+ "blocking_user" => user
+ }) do
+ if last_activity_id do
+ stream_out_participations(conversation.participations)
+ end
+ end
+ end
+
+ def stream_out_participations(_, _), do: :noop
+
def stream_out(activity) do
public = "https://www.w3.org/ns/activitystreams#Public"
if activity.data["type"] in ["Create", "Announce", "Delete"] do
- Pleroma.Web.Streamer.stream("user", activity)
- Pleroma.Web.Streamer.stream("list", activity)
-
- if Enum.member?(activity.data["to"], public) do
- Pleroma.Web.Streamer.stream("public", activity)
+ object = Object.normalize(activity)
+ # Do not stream out poll replies
+ unless object.data["type"] == "Answer" do
+ Pleroma.Web.Streamer.stream("user", activity)
+ Pleroma.Web.Streamer.stream("list", activity)
- if activity.local do
- Pleroma.Web.Streamer.stream("public:local", activity)
- end
+ if Enum.member?(activity.data["to"], public) do
+ Pleroma.Web.Streamer.stream("public", activity)
- if activity.data["type"] in ["Create"] do
- object = Object.normalize(activity)
+ if activity.local do
+ Pleroma.Web.Streamer.stream("public:local", activity)
+ end
- object.data
- |> Map.get("tag", [])
- |> Enum.filter(fn tag -> is_bitstring(tag) end)
- |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
+ if activity.data["type"] in ["Create"] do
+ object.data
+ |> Map.get("tag", [])
+ |> Enum.filter(fn tag -> is_bitstring(tag) end)
+ |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
- if object.data["attachment"] != [] do
- Pleroma.Web.Streamer.stream("public:media", activity)
+ if object.data["attachment"] != [] do
+ Pleroma.Web.Streamer.stream("public:media", activity)
- if activity.local do
- Pleroma.Web.Streamer.stream("public:local:media", activity)
+ if activity.local do
+ Pleroma.Web.Streamer.stream("public:local:media", activity)
+ end
end
end
+ else
+ # TODO: Write test, replace with visibility test
+ if !Enum.member?(activity.data["cc"] || [], public) &&
+ !Enum.member?(
+ activity.data["to"],
+ User.get_cached_by_ap_id(activity.data["actor"]).follower_address
+ ),
+ do: Pleroma.Web.Streamer.stream("direct", activity)
end
- else
- # TODO: Write test, replace with visibility test
- if !Enum.member?(activity.data["cc"] || [], public) &&
- !Enum.member?(
- activity.data["to"],
- User.get_cached_by_ap_id(activity.data["actor"]).follower_address
- ),
- do: Pleroma.Web.Streamer.stream("direct", activity)
end
end
end
"to" => to,
"deleted_activity_id" => activity && activity.id
},
- {:ok, activity} <- insert(data, local),
+ {:ok, activity} <- insert(data, local, false),
+ stream_out_participations(object, user),
_ <- decrease_replies_count_if_reply(object),
# Changing note count prior to enqueuing federation task in order to avoid
# race conditions on updating user.info
end
def block(blocker, blocked, activity_id \\ nil, local \\ true) do
- ap_config = Application.get_env(:pleroma, :activitypub)
- unfollow_blocked = Keyword.get(ap_config, :unfollow_blocked)
- outgoing_blocks = Keyword.get(ap_config, :outgoing_blocks)
+ outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
+ unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
- with true <- unfollow_blocked do
+ if unfollow_blocked do
follow_activity = fetch_latest_follow(blocker, blocked)
-
- if follow_activity do
- unfollow(blocker, blocked, nil, local)
- end
+ if follow_activity, do: unfollow(blocker, blocked, nil, local)
end
with true <- outgoing_blocks,
if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
from(activity in Activity)
+ |> maybe_preload_objects(opts)
|> restrict_blocked(opts)
- |> restrict_poll_replies(opts)
|> restrict_recipients(recipients, opts["user"])
|> where(
[activity],
^context
)
)
+ |> exclude_poll_votes(opts)
|> order_by([activity], desc: activity.id)
end
def fetch_activities_for_context(context, opts \\ %{}) do
context
|> fetch_activities_for_context_query(opts)
- |> Activity.with_preloaded_object()
|> Repo.all()
end
Pleroma.FlakeId.t() | nil
def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
context
- |> fetch_activities_for_context_query(opts)
+ |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
|> limit(1)
|> select([a], a.id)
|> Repo.one()
defp restrict_visibility(query, %{visibility: visibility})
when visibility in @valid_visibilities do
- query =
- from(
- a in query,
- where:
- fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
- )
-
- query
+ from(
+ a in query,
+ where:
+ fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
+ )
end
defp restrict_visibility(_query, %{visibility: visibility})
defp restrict_visibility(query, _visibility), do: query
- defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}) do
- query =
- from(
- a in query,
- where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
- )
+ defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
+ do: query
- query
+ defp restrict_thread_visibility(
+ query,
+ %{"user" => %User{info: %{skip_thread_containment: true}}},
+ _
+ ),
+ do: query
+
+ defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
+ from(
+ a in query,
+ where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
+ )
end
- defp restrict_thread_visibility(query, _), do: query
+ defp restrict_thread_visibility(query, _, _), do: query
def fetch_user_activities(user, reading_user, params \\ %{}) do
params =
defp restrict_tag(query, _), do: query
- defp restrict_to_cc(query, recipients_to, recipients_cc) do
- from(
- activity in query,
- where:
- fragment(
- "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
- activity.data,
- ^recipients_to,
- activity.data,
- ^recipients_cc
- )
- )
- end
-
defp restrict_recipients(query, [], _user), do: query
defp restrict_recipients(query, recipients, nil) do
defp restrict_muted_reblogs(query, _), do: query
- defp restrict_poll_replies(query, %{"include_poll_replies" => "true"}), do: query
+ defp exclude_poll_votes(query, %{"include_poll_votes" => "true"}), do: query
- defp restrict_poll_replies(query, _) do
+ defp exclude_poll_votes(query, _) do
if has_named_binding?(query, :object) do
- from([activity, object: o] in query, where: fragment("?->'name' is null", o.data))
+ from([activity, object: o] in query,
+ where: fragment("not(?->>'type' = ?)", o.data, "Answer")
+ )
else
query
end
|> Activity.with_preloaded_bookmark(opts["user"])
end
+ defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
+
+ defp maybe_set_thread_muted_field(query, opts) do
+ query
+ |> Activity.with_set_thread_muted_field(opts["user"])
+ end
+
defp maybe_order(query, %{order: :desc}) do
query
|> order_by(desc: :id)
def fetch_activities_query(recipients, opts \\ %{}) do
base_query = from(activity in Activity)
+ config = %{
+ skip_thread_containment: Config.get([:instance, :skip_thread_containment])
+ }
+
base_query
|> maybe_preload_objects(opts)
|> maybe_preload_bookmarks(opts)
+ |> maybe_set_thread_muted_field(opts)
|> maybe_order(opts)
|> restrict_recipients(recipients, opts["user"])
|> restrict_tag(opts)
|> restrict_muted(opts)
|> restrict_media(opts)
|> restrict_visibility(opts)
- |> restrict_thread_visibility(opts)
+ |> restrict_thread_visibility(opts, config)
|> restrict_replies(opts)
|> restrict_reblogs(opts)
|> restrict_pinned(opts)
|> restrict_muted_reblogs(opts)
|> Activity.restrict_deactivated_users()
- |> restrict_poll_replies(opts)
+ |> exclude_poll_votes(opts)
end
def fetch_activities(recipients, opts \\ %{}) do
|> Enum.reverse()
end
- def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
+ def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
+ from(activity in query,
+ where:
+ fragment("? && ?", activity.recipients, ^recipients) or
+ (fragment("? && ?", activity.recipients, ^recipients_with_public) and
+ "https://www.w3.org/ns/activitystreams#Public" in activity.recipients)
+ )
+ end
+
+ def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
fetch_activities_query([], opts)
- |> restrict_to_cc(recipients_to, recipients_cc)
+ |> fetch_activities_bounded_query(recipients, recipients_with_public)
|> Pagination.fetch_paginated(opts)
|> Enum.reverse()
end
end
end
- def user_data_from_user_object(data) do
+ defp object_to_user_data(data) do
avatar =
data["icon"]["url"] &&
%{
{:ok, user_data}
end
+ def user_data_from_user_object(data) do
+ with {:ok, data} <- MRF.filter(data),
+ {:ok, data} <- object_to_user_data(data) do
+ {:ok, data}
+ else
+ e -> {:error, e}
+ end
+ end
+
def fetch_and_prepare_user_from_ap_id(ap_id) do
- with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
- user_data_from_user_object(data)
+ with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
+ {:ok, data} <- user_data_from_user_object(data) do
+ {:ok, data}
else
e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
end