defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
alias Pleroma.Conversation
- alias Pleroma.Instances
alias Pleroma.Notification
alias Pleroma.Object
alias Pleroma.Object.Fetcher
alias Pleroma.User
alias Pleroma.Web.ActivityPub.MRF
alias Pleroma.Web.ActivityPub.Transmogrifier
- alias Pleroma.Web.Federator
alias Pleroma.Web.WebFinger
import Ecto.Query
require Logger
- @httpoison Application.get_env(:pleroma, :httpoison)
-
# For Announce activities, we filter the recipients based on following status for any actors
# that match actual users. See issue #164 for more information about why this is necessary.
defp get_recipients(%{"type" => "Announce"} = data) do
def decrease_replies_count_if_reply(_object), do: :noop
+ def increase_poll_votes_if_vote(%{
+ "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
+ "type" => "Create"
+ }) do
+ Object.increase_vote_count(reply_ap_id, name)
+ end
+
+ def increase_poll_votes_if_vote(_create_data), do: :noop
+
def insert(map, local \\ true, fake \\ false) when is_map(map) do
with nil <- Activity.normalize(map),
map <- lazy_put_activity_defaults(map, fake),
activity
end
- Task.start(fn ->
- Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
- end)
+ PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
Notification.create_notifications(activity)
{:ok, activity} <- insert(create_data, local, fake),
{:fake, false, activity} <- {:fake, fake, activity},
_ <- increase_replies_count_if_reply(create_data),
+ _ <- increase_poll_votes_if_vote(create_data),
# Changing note count prior to enqueuing federation task in order to avoid
# race conditions on updating user.info
{:ok, _actor} <- increase_note_count_if_public(actor, activity),
from(activity in Activity)
|> restrict_blocked(opts)
+ |> restrict_poll_replies(opts)
|> restrict_recipients(recipients, opts["user"])
|> where(
[activity],
defp restrict_visibility(query, %{visibility: visibility})
when is_list(visibility) do
if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
- from(
- a in query,
- where:
- fragment(
- "activity_visibility(?, ?, ?) = ANY (?)",
- a.actor,
- a.recipients,
- a.data,
- ^visibility
- )
- )
+ query =
+ from(
+ a in query,
+ where:
+ fragment(
+ "activity_visibility(?, ?, ?) = ANY (?)",
+ a.actor,
+ a.recipients,
+ a.data,
+ ^visibility
+ )
+ )
+
+ query
else
Logger.error("Could not restrict visibility to #{visibility}")
end
defp restrict_visibility(query, %{visibility: visibility})
when visibility in @valid_visibilities do
- 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)
+ )
+
+ query
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)
+ )
+
+ query
+ end
+
+ defp restrict_thread_visibility(query, _), do: query
+
def fetch_user_activities(user, reading_user, params \\ %{}) do
params =
params
defp restrict_type(query, _), do: query
+ defp restrict_state(query, %{"state" => state}) do
+ from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
+ end
+
+ defp restrict_state(query, _), do: query
+
defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
from(
activity in query,
blocks = info.blocks || []
domain_blocks = info.domain_blocks || []
+ query =
+ if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
+
from(
- activity in query,
+ [activity, object: o] in query,
where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
where: fragment("not (? && ?)", activity.recipients, ^blocks),
where:
activity.data,
^blocks
),
- where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
+ where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
+ where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
)
end
defp restrict_muted_reblogs(query, _), do: query
+ defp restrict_poll_replies(query, %{"include_poll_replies" => "true"}), do: query
+
+ defp restrict_poll_replies(query, _) do
+ if has_named_binding?(query, :object) do
+ from([activity, object: o] in query, where: fragment("?->'name' is null", o.data))
+ else
+ query
+ end
+ end
+
defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
defp maybe_preload_objects(query, _) do
|> restrict_local(opts)
|> restrict_actor(opts)
|> restrict_type(opts)
+ |> restrict_state(opts)
|> restrict_favorited_by(opts)
|> restrict_blocked(opts)
|> restrict_muted(opts)
|> restrict_media(opts)
|> restrict_visibility(opts)
+ |> restrict_thread_visibility(opts)
|> restrict_replies(opts)
|> restrict_reblogs(opts)
|> restrict_pinned(opts)
|> restrict_muted_reblogs(opts)
+ |> Activity.restrict_deactivated_users()
+ |> restrict_poll_replies(opts)
end
def fetch_activities(recipients, opts \\ %{}) do
end
end
- def should_federate?(inbox, public) do
- if public do
- true
- else
- inbox_info = URI.parse(inbox)
- !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
- end
- end
-
- def publish(actor, activity) do
- remote_followers =
- if actor.follower_address in activity.recipients do
- {:ok, followers} = User.get_followers(actor)
- followers |> Enum.filter(&(!&1.local))
- else
- []
- end
-
- public = is_public?(activity)
-
- {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
- json = Jason.encode!(data)
-
- (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
- |> Enum.filter(fn user -> User.ap_enabled?(user) end)
- |> Enum.map(fn %{info: %{source_data: data}} ->
- (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
- end)
- |> Enum.uniq()
- |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
- |> Instances.filter_reachable()
- |> Enum.each(fn {inbox, unreachable_since} ->
- Federator.publish_single_ap(%{
- inbox: inbox,
- json: json,
- actor: actor,
- id: activity.data["id"],
- unreachable_since: unreachable_since
- })
- end)
- end
-
- def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do
- Logger.info("Federating #{id} to #{inbox}")
- host = URI.parse(inbox).host
-
- digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
-
- date =
- NaiveDateTime.utc_now()
- |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
-
- signature =
- Pleroma.Web.HTTPSignatures.sign(actor, %{
- host: host,
- "content-length": byte_size(json),
- digest: digest,
- date: date
- })
-
- with {:ok, %{status: code}} when code in 200..299 <-
- result =
- @httpoison.post(
- inbox,
- json,
- [
- {"Content-Type", "application/activity+json"},
- {"Date", date},
- {"signature", signature},
- {"digest", digest}
- ]
- ) do
- if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
- do: Instances.set_reachable(inbox)
-
- result
- else
- {_post_result, response} ->
- unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
- {:error, response}
- end
- end
-
# filter out broken threads
def contain_broken_threads(%Activity{} = activity, %User{} = user) do
entire_thread_visible_for_user?(activity, user)
contain_broken_threads(activity, user)
end
- # do post-processing on a timeline
- def contain_timeline(timeline, user) do
- timeline
- |> Enum.filter(fn activity ->
- contain_activity(activity, user)
- end)
- end
-
def fetch_direct_messages_query do
Activity
|> restrict_type(%{"type" => "Create"})
|> restrict_visibility(%{visibility: "direct"})
+ |> order_by([activity], asc: activity.id)
end
end