require Logger
require Pleroma.Constants
+ @behaviour Pleroma.Web.ActivityPub.ActivityPub.Persisting
+ @behaviour Pleroma.Web.ActivityPub.ActivityPub.Streaming
+
defp get_recipients(%{"type" => "Create"} = data) do
to = Map.get(data, "to", [])
cc = Map.get(data, "cc", [])
defp increase_replies_count_if_reply(_create_data), do: :noop
@object_types ~w[ChatMessage Question Answer Audio Video Event Article]
- @spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
+ @impl true
def persist(%{"type" => type} = object, meta) when type in @object_types do
with {:ok, object} <- Object.create(object) do
{:ok, object, meta}
end
end
+ @impl true
def persist(object, meta) do
with local <- Keyword.fetch!(meta, :local),
{recipients, _, _} <- get_recipients(object),
# Splice in the child object if we have one.
activity = Maps.put_if_present(activity, :object, object)
- BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
+ ConcurrentLimiter.limit(Pleroma.Web.RichMedia.Helpers, fn ->
+ Task.start(fn -> Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity) end)
+ end)
{:ok, activity}
else
Streamer.stream("participation", participations)
end
+ @impl true
def stream_out_participations(%Object{data: %{"context" => context}}, user) do
with %Conversation{} = conversation <- Conversation.get_for_ap_id(context) do
conversation = Repo.preload(conversation, :participations)
end
end
+ @impl true
def stream_out_participations(_, _), do: :noop
+ @impl true
def stream_out(%Activity{data: %{"type" => data_type}} = activity)
when data_type in ["Create", "Announce", "Delete"] do
activity
|> Streamer.stream(activity)
end
+ @impl true
def stream_out(_activity) do
:noop
end
end
@spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
- def flag(
- %{
- actor: actor,
- context: _context,
- account: account,
- statuses: statuses,
- content: content
- } = params
- ) do
+ def flag(params) do
+ with {:ok, result} <- Repo.transaction(fn -> do_flag(params) end) do
+ result
+ end
+ end
+
+ defp do_flag(
+ %{
+ actor: actor,
+ context: _context,
+ account: account,
+ statuses: statuses,
+ content: content
+ } = params
+ ) do
# only accept false as false value
local = !(params[:local] == false)
forward = !(params[:forward] == false)
{:ok, activity} <- insert(flag_data, local),
{:ok, stripped_activity} <- strip_report_status_data(activity),
_ <- notify_and_stream(activity),
- :ok <- maybe_federate(stripped_activity) do
+ :ok <-
+ maybe_federate(stripped_activity) do
User.all_superusers()
|> Enum.filter(fn user -> not is_nil(user.email) end)
|> Enum.each(fn superuser ->
end)
{:ok, activity}
+ else
+ {:error, error} -> Repo.rollback(error)
end
end
|> Map.put(:muting_user, reading_user)
end
+ pagination_type = Map.get(params, :pagination_type) || :keyset
+
%{
godmode: params[:godmode],
reading_user: reading_user
}
|> user_activities_recipients()
- |> fetch_activities(params)
+ |> fetch_activities(params, pagination_type)
|> Enum.reverse()
end
where:
fragment(
"""
- ?->>'type' != 'Create' -- This isn't a Create
+ ?->>'type' != 'Create' -- This isn't a Create
OR ?->>'inReplyTo' is null -- this isn't a reply
- OR ? && array_remove(?, ?) -- The recipient is us or one of our friends,
- -- unless they are the author (because authors
+ OR ? && array_remove(?, ?) -- The recipient is us or one of our friends,
+ -- unless they are the author (because authors
-- are also part of the recipients). This leads
-- to a bug that self-replies by friends won't
-- show up.
defp restrict_muted_reblogs(query, _), do: query
- defp restrict_instance(query, %{instance: instance}) do
- users =
- from(
- u in User,
- select: u.ap_id,
- where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
- )
- |> Repo.all()
-
- from(activity in query, where: activity.actor in ^users)
+ defp restrict_instance(query, %{instance: instance}) when is_binary(instance) do
+ from(
+ activity in query,
+ where: fragment("split_part(actor::text, '/'::text, 3) = ?", ^instance)
+ )
end
defp restrict_instance(query, _), do: query
def fetch_follow_information_for_user(user) do
with {:ok, following_data} <-
- Fetcher.fetch_and_contain_remote_object_from_id(user.following_address,
- force_http: true
- ),
+ Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
{:ok, hide_follows} <- collection_private(following_data),
{:ok, followers_data} <-
- Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address, force_http: true),
+ Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
{:ok, hide_followers} <- collection_private(followers_data) do
{:ok,
%{
end
end
- def fetch_and_prepare_user_from_ap_id(ap_id, opts \\ []) do
- with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id, opts),
+ def fetch_and_prepare_user_from_ap_id(ap_id) do
+ with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
{:ok, data} <- user_data_from_user_object(data) do
{:ok, maybe_update_follow_information(data)}
else
+ # If this has been deleted, only log a debug and not an error
+ {:error, "Object has been deleted" = e} ->
+ Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
+ {:error, e}
+
{:error, {:reject, reason} = e} ->
Logger.info("Rejected user #{ap_id}: #{inspect(reason)}")
{:error, e}
end
end
- def make_user_from_ap_id(ap_id, opts \\ []) do
+ def make_user_from_ap_id(ap_id) do
user = User.get_cached_by_ap_id(ap_id)
if user && !User.ap_enabled?(user) do
Transmogrifier.upgrade_user_from_ap_id(ap_id)
else
- with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, opts) do
+ with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
if user do
user
|> User.remote_user_changeset(data)
projects / akkoma / blobdiff