X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Factivity_pub%2Factivity_pub.ex;h=54d1a2350ea670c12cd37752150f09f869d49401;hb=4134abef63e1165f5701741c1012e64cb908654c;hp=d17c892a7e2b6d6600220e37a2812176354d32fe;hpb=18a91d85e59eb67a2d58a28aae732c34b75ebe97;p=akkoma diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index d17c892a7..54d1a2350 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -32,6 +32,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do require Logger require Pleroma.Constants + @behaviour Pleroma.Web.ActivityPub.ActivityPub.Persisting + defp get_recipients(%{"type" => "Create"} = data) do to = Map.get(data, "to", []) cc = Map.get(data, "cc", []) @@ -85,13 +87,14 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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), @@ -123,7 +126,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do # 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 @@ -332,15 +337,21 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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) @@ -358,7 +369,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do {: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 -> @@ -368,6 +380,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end) {:ok, activity} + else + {:error, error} -> Repo.rollback(error) end end @@ -649,33 +663,41 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do defp restrict_since(query, _), do: query defp restrict_tag_reject(_query, %{tag_reject: _tag_reject, skip_preload: true}) do - raise "Can't use the child object without preloading!" + raise_on_missing_preload() end - defp restrict_tag_reject(query, %{tag_reject: [_ | _] = tag_reject}) do + defp restrict_tag_reject(query, %{tag_reject: tag_reject}) when is_list(tag_reject) do from( [_activity, object] in query, where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject) ) end + defp restrict_tag_reject(query, %{tag_reject: tag_reject}) when is_binary(tag_reject) do + restrict_tag_reject(query, %{tag_reject: [tag_reject]}) + end + defp restrict_tag_reject(query, _), do: query defp restrict_tag_all(_query, %{tag_all: _tag_all, skip_preload: true}) do - raise "Can't use the child object without preloading!" + raise_on_missing_preload() end - defp restrict_tag_all(query, %{tag_all: [_ | _] = tag_all}) do + defp restrict_tag_all(query, %{tag_all: tag_all}) when is_list(tag_all) do from( [_activity, object] in query, where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all) ) end + defp restrict_tag_all(query, %{tag_all: tag}) when is_binary(tag) do + restrict_tag(query, %{tag: tag}) + end + defp restrict_tag_all(query, _), do: query defp restrict_tag(_query, %{tag: _tag, skip_preload: true}) do - raise "Can't use the child object without preloading!" + raise_on_missing_preload() end defp restrict_tag(query, %{tag: tag}) when is_list(tag) do @@ -686,13 +708,79 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end defp restrict_tag(query, %{tag: tag}) when is_binary(tag) do + restrict_tag(query, %{tag: [tag]}) + end + + defp restrict_tag(query, _), do: query + + defp restrict_hashtag_reject_any(_query, %{tag_reject: _tag_reject, skip_preload: true}) do + raise_on_missing_preload() + end + + defp restrict_hashtag_reject_any(query, %{tag_reject: tags_reject}) when is_list(tags_reject) do + if has_named_binding?(query, :thread_mute) do + from( + [activity, object, thread_mute] in query, + group_by: [activity.id, object.id, thread_mute.id] + ) + else + from( + [activity, object] in query, + group_by: [activity.id, object.id] + ) + end + |> join(:left, [_activity, object], hashtag in assoc(object, :hashtags), as: :hashtag) + |> having( + [hashtag: hashtag], + fragment("not(array_agg(?) && (?))", hashtag.name, ^tags_reject) + ) + end + + defp restrict_hashtag_reject_any(query, %{tag_reject: tag_reject}) when is_binary(tag_reject) do + restrict_hashtag_reject_any(query, %{tag_reject: [tag_reject]}) + end + + defp restrict_hashtag_reject_any(query, _), do: query + + defp restrict_hashtag_all(_query, %{tag_all: _tag, skip_preload: true}) do + raise_on_missing_preload() + end + + defp restrict_hashtag_all(query, %{tag_all: tags}) when is_list(tags) do + Enum.reduce( + tags, + query, + fn tag, acc -> restrict_hashtag_any(acc, %{tag: tag}) end + ) + end + + defp restrict_hashtag_all(query, %{tag_all: tag}) when is_binary(tag) do + restrict_hashtag_any(query, %{tag: tag}) + end + + defp restrict_hashtag_all(query, _), do: query + + defp restrict_hashtag_any(_query, %{tag: _tag, skip_preload: true}) do + raise_on_missing_preload() + end + + defp restrict_hashtag_any(query, %{tag: tags}) when is_list(tags) do from( [_activity, object] in query, - where: fragment("(?)->'tag' \\? (?)", object.data, ^tag) + join: hashtag in assoc(object, :hashtags), + where: hashtag.name in ^tags ) end - defp restrict_tag(query, _), do: query + defp restrict_hashtag_any(query, %{tag: tag}) when is_binary(tag) do + restrict_hashtag_any(query, %{tag: [tag]}) + end + + defp restrict_hashtag_any(query, _), do: query + + defp raise_on_missing_preload do + raise "Can't use the child object without preloading!" + end defp restrict_recipients(query, [], _user), do: query @@ -791,10 +879,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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. @@ -937,16 +1025,11 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -1082,40 +1165,51 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do skip_thread_containment: Config.get([:instance, :skip_thread_containment]) } - Activity - |> maybe_preload_objects(opts) - |> maybe_preload_bookmarks(opts) - |> maybe_preload_report_notes(opts) - |> maybe_set_thread_muted_field(opts) - |> maybe_order(opts) - |> restrict_recipients(recipients, opts[:user]) - |> restrict_replies(opts) - |> restrict_tag(opts) - |> restrict_tag_reject(opts) - |> restrict_tag_all(opts) - |> restrict_since(opts) - |> restrict_local(opts) - |> restrict_actor(opts) - |> restrict_type(opts) - |> restrict_state(opts) - |> 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_reblogs(opts) - |> restrict_pinned(opts) - |> 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) - |> exclude_invisible_actors(opts) - |> exclude_visibility(opts) + query = + Activity + |> distinct([a], true) + |> maybe_preload_objects(opts) + |> maybe_preload_bookmarks(opts) + |> maybe_preload_report_notes(opts) + |> maybe_set_thread_muted_field(opts) + |> maybe_order(opts) + |> restrict_recipients(recipients, opts[:user]) + |> restrict_replies(opts) + |> restrict_since(opts) + |> restrict_local(opts) + |> restrict_actor(opts) + |> restrict_type(opts) + |> restrict_state(opts) + |> 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_reblogs(opts) + |> restrict_pinned(opts) + |> 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) + |> exclude_invisible_actors(opts) + |> exclude_visibility(opts) + + if Config.get([:instance, :improved_hashtag_timeline]) do + query + |> restrict_hashtag_any(opts) + |> restrict_hashtag_all(opts) + |> restrict_hashtag_reject_any(opts) + else + query + |> restrict_tag(opts) + |> restrict_tag_reject(opts) + |> restrict_tag_all(opts) + end end def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do @@ -1294,12 +1388,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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, %{ @@ -1373,11 +1465,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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} @@ -1416,13 +1509,13 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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)