X-Git-Url: https://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Factivity_pub%2Factivity_pub.ex;h=8c2610eeb377ac192105d0b042ca73f8f96e740a;hb=20a269ed694127ca59ece9c2e4865a627b4daad3;hp=d17c892a7e2b6d6600220e37a2812176354d32fe;hpb=2fd384e0a28c2e8259ba1afccb2495f4a2399a58;p=akkoma diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index d17c892a7..8c2610eeb 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -32,6 +32,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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", []) @@ -85,13 +88,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 +127,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 @@ -219,6 +225,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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) @@ -235,8 +242,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do 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 @@ -244,6 +253,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> Streamer.stream(activity) end + @impl true def stream_out(_activity) do :noop end @@ -332,15 +342,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 +374,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 +385,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end) {:ok, activity} + else + {:error, error} -> Repo.rollback(error) end end @@ -589,12 +608,14 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do |> 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 @@ -791,10 +812,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 +958,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 @@ -1294,12 +1310,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 +1387,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 +1431,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)