defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
alias Pleroma.Activity.Ir.Topics
- alias Pleroma.ActivityExpiration
alias Pleroma.Config
alias Pleroma.Constants
alias Pleroma.Conversation
alias Pleroma.Conversation.Participation
+ alias Pleroma.Filter
alias Pleroma.Maps
alias Pleroma.Notification
alias Pleroma.Object
defp check_remote_limit(_), do: true
- defp increase_note_count_if_public(actor, object) do
+ def increase_note_count_if_public(actor, object) do
if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
end
defp increase_replies_count_if_reply(_create_data), do: :noop
- defp increase_poll_votes_if_vote(%{
- "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
- "type" => "Create",
- "actor" => actor
- }) do
- Object.increase_vote_count(reply_ap_id, name, actor)
- end
-
- defp increase_poll_votes_if_vote(_create_data), do: :noop
-
- @object_types ["ChatMessage"]
+ @object_types ~w[ChatMessage Question Answer Audio Event]
@spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
def persist(%{"type" => type} = object, meta) when type in @object_types do
with {:ok, object} <- Object.create(object) do
local: local,
recipients: recipients,
actor: object["actor"]
- }) do
+ }),
+ # TODO: add tests for expired activities, when Note type will be supported in new pipeline
+ {:ok, _} <- maybe_create_activity_expiration(activity) do
{:ok, activity, meta}
end
end
def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
with nil <- Activity.normalize(map),
map <- lazy_put_activity_defaults(map, fake),
- true <- bypass_actor_check || check_actor_is_active(map["actor"]),
- {_, true} <- {:remote_limit_error, check_remote_limit(map)},
+ {_, true} <- {:actor_check, bypass_actor_check || check_actor_is_active(map["actor"])},
+ {_, true} <- {:remote_limit_pass, check_remote_limit(map)},
{:ok, map} <- MRF.filter(map),
{recipients, _, _} = get_recipients(map),
{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
{:containment, :ok} <- {:containment, Containment.contain_child(map)},
- {:ok, map, object} <- insert_full_object(map) do
- {:ok, activity} =
- %Activity{
- data: map,
- local: local,
- actor: map["actor"],
- recipients: recipients
- }
- |> Repo.insert()
- |> maybe_create_activity_expiration()
-
+ {:ok, map, object} <- insert_full_object(map),
+ {:ok, activity} <- insert_activity_with_expiration(map, local, recipients) do
# Splice in the child object if we have one.
activity = Maps.put_if_present(activity, :object, object)
%Activity{} = activity ->
{:ok, activity}
+ {:actor_check, _} ->
+ {:error, false}
+
+ {:containment, _} = error ->
+ error
+
+ {:error, _} = error ->
+ error
+
{:fake, true, map, recipients} ->
activity = %Activity{
data: map,
Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
{:ok, activity}
- error ->
- {:error, error}
+ {:remote_limit_pass, _} ->
+ {:error, :remote_limit}
+
+ {:reject, reason} ->
+ {:error, reason}
+ end
+ end
+
+ defp insert_activity_with_expiration(data, local, recipients) do
+ struct = %Activity{
+ data: data,
+ local: local,
+ actor: data["actor"],
+ recipients: recipients
+ }
+
+ with {:ok, activity} <- Repo.insert(struct) do
+ maybe_create_activity_expiration(activity)
end
end
stream_out_participations(participations)
end
- defp maybe_create_activity_expiration({:ok, %{data: %{"expires_at" => expires_at}} = activity}) do
- with {:ok, _} <- ActivityExpiration.create(activity, expires_at) do
+ defp maybe_create_activity_expiration(
+ %{data: %{"expires_at" => %DateTime{} = expires_at}} = activity
+ ) do
+ with {:ok, _job} <-
+ Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+ activity_id: activity.id,
+ expires_at: expires_at
+ }) do
{:ok, activity}
end
end
- defp maybe_create_activity_expiration(result), do: result
+ defp maybe_create_activity_expiration(activity), do: {:ok, activity}
defp create_or_bump_conversation(activity, actor) do
with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
with {: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),
{:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
{:ok, _actor} <- increase_note_count_if_public(actor, activity),
_ <- notify_and_stream(activity),
end
end
- @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
- def accept(params) do
- accept_or_reject("Accept", params)
- end
-
- @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
- def reject(params) do
- accept_or_reject("Reject", params)
- end
-
- @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
- defp accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
- local = Map.get(params, :local, true)
- activity_id = Map.get(params, :activity_id, nil)
-
- data =
- %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
- |> Maps.put_if_present("id", activity_id)
-
- with {:ok, activity} <- insert(data, local),
- _ <- notify_and_stream(activity),
- :ok <- maybe_federate(activity) do
- {:ok, activity}
- end
- end
-
- @spec follow(User.t(), User.t(), String.t() | nil, boolean(), keyword()) ::
- {:ok, Activity.t()} | {:error, any()}
- def follow(follower, followed, activity_id \\ nil, local \\ true, opts \\ []) do
- with {:ok, result} <-
- Repo.transaction(fn -> do_follow(follower, followed, activity_id, local, opts) end) do
- result
- end
- end
-
- defp do_follow(follower, followed, activity_id, local, opts) do
- skip_notify_and_stream = Keyword.get(opts, :skip_notify_and_stream, false)
- data = make_follow_data(follower, followed, activity_id)
-
- with {:ok, activity} <- insert(data, local),
- _ <- skip_notify_and_stream || notify_and_stream(activity),
- :ok <- maybe_federate(activity) do
- {:ok, activity}
- else
- {:error, error} -> Repo.rollback(error)
- end
- end
-
@spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
{:ok, Activity.t()} | nil | {:error, any()}
def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
|> maybe_set_thread_muted_field(opts)
|> restrict_blocked(opts)
|> restrict_recipients(recipients, opts[:user])
+ |> restrict_filtered(opts)
|> where(
[activity],
fragment(
defp restrict_instance(query, _), do: query
+ defp restrict_filtered(query, %{user: %User{} = user}) do
+ case Filter.compose_regex(user) do
+ nil ->
+ query
+
+ regex ->
+ from([activity, object] in query,
+ where:
+ fragment("not(?->>'content' ~* ?)", object.data, ^regex) or
+ activity.actor == ^user.ap_id
+ )
+ end
+ end
+
+ defp restrict_filtered(query, %{blocking_user: %User{} = user}) do
+ restrict_filtered(query, %{user: user})
+ end
+
+ defp restrict_filtered(query, _), do: query
+
defp exclude_poll_votes(query, %{include_poll_votes: true}), do: query
defp exclude_poll_votes(query, _) do
|> restrict_favorited_by(opts)
|> restrict_blocked(restrict_blocked_opts)
|> restrict_muted(restrict_muted_opts)
+ |> restrict_filtered(opts)
|> restrict_media(opts)
|> restrict_visibility(opts)
|> restrict_thread_visibility(opts, config)
|> restrict_muted_reblogs(restrict_muted_reblogs_opts)
|> restrict_instance(opts)
|> restrict_announce_object_actor(opts)
+ |> restrict_filtered(opts)
|> Activity.restrict_deactivated_users()
|> exclude_poll_votes(opts)
|> exclude_chat_messages(opts)
name: data["name"],
follower_address: data["followers"],
following_address: data["following"],
- bio: data["summary"],
+ bio: data["summary"] || "",
actor_type: actor_type,
also_known_as: Map.get(data, "alsoKnownAs", []),
public_key: public_key,
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}
+
{:error, e} ->
Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
{:error, e}
end
end
- def maybe_handle_clashing_nickname(nickname) do
- with %User{} = old_user <- User.get_by_nickname(nickname) do
- Logger.info("Found an old user for #{nickname}, ap id is #{old_user.ap_id}, renaming.")
+ def maybe_handle_clashing_nickname(data) do
+ with nickname when is_binary(nickname) <- data[:nickname],
+ %User{} = old_user <- User.get_by_nickname(nickname),
+ {_, false} <- {:ap_id_comparison, data[:ap_id] == old_user.ap_id} do
+ Logger.info(
+ "Found an old user for #{nickname}, the old ap id is #{old_user.ap_id}, new one is #{
+ data[:ap_id]
+ }, renaming."
+ )
old_user
|> User.remote_user_changeset(%{nickname: "#{old_user.id}.#{old_user.nickname}"})
|> User.update_and_set_cache()
+ else
+ {:ap_id_comparison, true} ->
+ Logger.info(
+ "Found an old user for #{data[:nickname]}, but the ap id #{data[:ap_id]} is the same as the new user. Race condition? Not changing anything."
+ )
+
+ _ ->
+ nil
end
end
|> User.remote_user_changeset(data)
|> User.update_and_set_cache()
else
- maybe_handle_clashing_nickname(data[:nickname])
+ maybe_handle_clashing_nickname(data)
data
|> User.remote_user_changeset()
projects / akkoma / blobdiff