Revert "Merge branch 'streamer-refactoring' into 'develop'"
[akkoma] / lib / pleroma / web / activity_pub / activity_pub.ex
index df4155d211f00d45931411d84a87843e9528084a..41f6a0f1f79fde06025d6c14adfe71cb4f4f6f91 100644 (file)
@@ -8,6 +8,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Conversation
   alias Pleroma.Notification
   alias Pleroma.Object
+  alias Pleroma.Object.Containment
   alias Pleroma.Object.Fetcher
   alias Pleroma.Pagination
   alias Pleroma.Repo
@@ -16,29 +17,28 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.WebFinger
+  alias Pleroma.Workers.BackgroundWorker
 
   import Ecto.Query
   import Pleroma.Web.ActivityPub.Utils
   import Pleroma.Web.ActivityPub.Visibility
 
   require Logger
+  require Pleroma.Constants
 
   # 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
-    to = data["to"] || []
-    cc = data["cc"] || []
+    to = Map.get(data, "to", [])
+    cc = Map.get(data, "cc", [])
+    bcc = Map.get(data, "bcc", [])
     actor = User.get_cached_by_ap_id(data["actor"])
 
     recipients =
-      (to ++ cc)
-      |> Enum.filter(fn recipient ->
+      Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
         case User.get_cached_by_ap_id(recipient) do
-          nil ->
-            true
-
-          user ->
-            User.following?(user, actor)
+          nil -> true
+          user -> User.following?(user, actor)
         end
       end)
 
@@ -46,17 +46,19 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   end
 
   defp get_recipients(%{"type" => "Create"} = data) do
-    to = data["to"] || []
-    cc = data["cc"] || []
-    actor = data["actor"] || []
-    recipients = (to ++ cc ++ [actor]) |> Enum.uniq()
+    to = Map.get(data, "to", [])
+    cc = Map.get(data, "cc", [])
+    bcc = Map.get(data, "bcc", [])
+    actor = Map.get(data, "actor", [])
+    recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
     {recipients, to, cc}
   end
 
   defp get_recipients(data) do
-    to = data["to"] || []
-    cc = data["cc"] || []
-    recipients = to ++ cc
+    to = Map.get(data, "to", [])
+    cc = Map.get(data, "cc", [])
+    bcc = Map.get(data, "bcc", [])
+    recipients = Enum.concat([to, cc, bcc])
     {recipients, to, cc}
   end
 
@@ -64,12 +66,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     if not is_nil(actor) do
       with user <- User.get_cached_by_ap_id(actor),
            false <- user.info.deactivated do
-        :ok
+        true
       else
-        _e -> :reject
+        _e -> false
       end
     else
-      :ok
+      true
     end
   end
 
@@ -118,14 +120,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   def increase_poll_votes_if_vote(_create_data), do: :noop
 
-  def insert(map, local \\ true, fake \\ false) when is_map(map) do
+  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),
-         :ok <- check_actor_is_active(map["actor"]),
+         true <- bypass_actor_check || check_actor_is_active(map["actor"]),
          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
          {:ok, map} <- MRF.filter(map),
          {recipients, _, _} = get_recipients(map),
          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
+         :ok <- Containment.contain_child(map),
          {:ok, map, object} <- insert_full_object(map) do
       {:ok, activity} =
         Repo.insert(%Activity{
@@ -137,13 +140,13 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
       # Splice in the child object if we have one.
       activity =
-        if !is_nil(object) do
+        if not is_nil(object) do
           Map.put(activity, :object, object)
         else
           activity
         end
 
-      PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
+      BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
 
       Notification.create_notifications(activity)
 
@@ -206,8 +209,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   def stream_out_participations(_, _), do: :noop
 
   def stream_out(activity) do
-    public = "https://www.w3.org/ns/activitystreams#Public"
-
     if activity.data["type"] in ["Create", "Announce", "Delete"] do
       object = Object.normalize(activity)
       # Do not stream out poll replies
@@ -215,7 +216,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
         Pleroma.Web.Streamer.stream("user", activity)
         Pleroma.Web.Streamer.stream("list", activity)
 
-        if Enum.member?(activity.data["to"], public) do
+        if get_visibility(activity) == "public" do
           Pleroma.Web.Streamer.stream("public", activity)
 
           if activity.local do
@@ -237,13 +238,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
             end
           end
         else
-          # TODO: Write test, replace with visibility test
-          if !Enum.member?(activity.data["cc"] || [], public) &&
-               !Enum.member?(
-                 activity.data["to"],
-                 User.get_cached_by_ap_id(activity.data["actor"]).follower_address
-               ),
-             do: Pleroma.Web.Streamer.stream("direct", activity)
+          if get_visibility(activity) == "direct",
+            do: Pleroma.Web.Streamer.stream("direct", activity)
         end
       end
     end
@@ -272,6 +268,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     else
       {:fake, true, activity} ->
         {:ok, activity}
+
+      {:error, message} ->
+        {:error, message}
     end
   end
 
@@ -333,12 +332,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
-  def unlike(
-        %User{} = actor,
-        %Object{} = object,
-        activity_id \\ nil,
-        local \\ true
-      ) do
+  def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
          {:ok, unlike_activity} <- insert(unlike_data, local),
@@ -390,7 +384,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   def follow(follower, followed, activity_id \\ nil, local \\ true) do
     with data <- make_follow_data(follower, followed, activity_id),
          {:ok, activity} <- insert(data, local),
-         :ok <- maybe_federate(activity) do
+         :ok <- maybe_federate(activity),
+         _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
       {:ok, activity}
     end
   end
@@ -412,7 +407,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
            "actor" => ap_id,
            "object" => %{"type" => "Person", "id" => ap_id}
          },
-         {:ok, activity} <- insert(data, true, true),
+         {:ok, activity} <- insert(data, true, true, true),
          :ok <- maybe_federate(activity) do
       {:ok, user}
     end
@@ -513,13 +508,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   end
 
   defp fetch_activities_for_context_query(context, opts) do
-    public = ["https://www.w3.org/ns/activitystreams#Public"]
+    public = [Pleroma.Constants.as_public()]
 
     recipients =
       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 
     from(activity in Activity)
     |> maybe_preload_objects(opts)
+    |> maybe_preload_bookmarks(opts)
+    |> maybe_set_thread_muted_field(opts)
     |> restrict_blocked(opts)
     |> restrict_recipients(recipients, opts["user"])
     |> where(
@@ -533,6 +530,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       )
     )
     |> exclude_poll_votes(opts)
+    |> exclude_id(opts)
     |> order_by([activity], desc: activity.id)
   end
 
@@ -554,7 +552,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   end
 
   def fetch_public_activities(opts \\ %{}) do
-    q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
+    q = fetch_activities_query([Pleroma.Constants.as_public()], opts)
 
     q
     |> restrict_unlisted()
@@ -625,22 +623,33 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     params =
       params
       |> Map.put("type", ["Create", "Announce"])
+      |> Map.put("user", reading_user)
       |> Map.put("actor_id", user.ap_id)
       |> Map.put("whole_db", true)
       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 
     recipients =
-      if reading_user do
-        ["https://www.w3.org/ns/activitystreams#Public"] ++
-          [reading_user.ap_id | reading_user.following]
-      else
-        ["https://www.w3.org/ns/activitystreams#Public"]
-      end
+      user_activities_recipients(%{
+        "godmode" => params["godmode"],
+        "reading_user" => reading_user
+      })
 
     fetch_activities(recipients, params)
     |> Enum.reverse()
   end
 
+  defp user_activities_recipients(%{"godmode" => true}) do
+    []
+  end
+
+  defp user_activities_recipients(%{"reading_user" => reading_user}) do
+    if reading_user do
+      [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
+    else
+      [Pleroma.Constants.as_public()]
+    end
+  end
+
   defp restrict_since(query, %{"since_id" => ""}), do: query
 
   defp restrict_since(query, %{"since_id" => since_id}) do
@@ -741,8 +750,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
     from(
-      activity in query,
-      where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
+      [_activity, object] in query,
+      where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
     )
   end
 
@@ -778,14 +787,20 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 
-  defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
+  defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
     mutes = info.mutes
 
-    from(
-      activity in query,
-      where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
-      where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
-    )
+    query =
+      from([activity] in query,
+        where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
+        where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
+      )
+
+    unless opts["skip_preload"] do
+      from([thread_mute: tm] in query, where: is_nil(tm.user_id))
+    else
+      query
+    end
   end
 
   defp restrict_muted(query, _), do: query
@@ -822,7 +837,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
         fragment(
           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
           activity.data,
-          ^["https://www.w3.org/ns/activitystreams#Public"]
+          ^[Pleroma.Constants.as_public()]
         )
     )
   end
@@ -862,6 +877,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
+  defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
+    from(activity in query, where: activity.id != ^id)
+  end
+
+  defp exclude_id(query, _), do: query
+
   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 
   defp maybe_preload_objects(query, _) do
@@ -880,7 +901,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp maybe_set_thread_muted_field(query, opts) do
     query
-    |> Activity.with_set_thread_muted_field(opts["user"])
+    |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
   end
 
   defp maybe_order(query, %{order: :desc}) do
@@ -896,13 +917,11 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   defp maybe_order(query, _), do: query
 
   def fetch_activities_query(recipients, opts \\ %{}) do
-    base_query = from(activity in Activity)
-
     config = %{
       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
     }
 
-    base_query
+    Activity
     |> maybe_preload_objects(opts)
     |> maybe_preload_bookmarks(opts)
     |> maybe_set_thread_muted_field(opts)
@@ -931,17 +950,37 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   end
 
   def fetch_activities(recipients, opts \\ %{}) do
-    fetch_activities_query(recipients, opts)
+    list_memberships = Pleroma.List.memberships(opts["user"])
+
+    fetch_activities_query(recipients ++ list_memberships, opts)
     |> Pagination.fetch_paginated(opts)
     |> Enum.reverse()
+    |> maybe_update_cc(list_memberships, opts["user"])
   end
 
+  defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
+       when is_list(list_memberships) and length(list_memberships) > 0 do
+    Enum.map(activities, fn
+      %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
+        if Enum.any?(bcc, &(&1 in list_memberships)) do
+          update_in(activity.data["cc"], &[user_ap_id | &1])
+        else
+          activity
+        end
+
+      activity ->
+        activity
+    end)
+  end
+
+  defp maybe_update_cc(activities, _, _), do: activities
+
   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
     from(activity in query,
       where:
         fragment("? && ?", activity.recipients, ^recipients) or
           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
-             "https://www.w3.org/ns/activitystreams#Public" in activity.recipients)
+             ^Pleroma.Constants.as_public() in activity.recipients)
     )
   end
 
@@ -980,6 +1019,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
           "url" => [%{"href" => data["image"]["url"]}]
         }
 
+    fields =
+      data
+      |> Map.get("attachment", [])
+      |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
+      |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
+
     locked = data["manuallyApprovesFollowers"] || false
     data = Transmogrifier.maybe_fix_user_object(data)
 
@@ -989,6 +1034,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
         ap_enabled: true,
         source_data: data,
         banner: banner,
+        fields: fields,
         locked: locked
       },
       avatar: avatar,
@@ -1022,15 +1068,13 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
          followers_count when is_integer(followers_count) <- followers_data["totalItems"],
          {:ok, hide_followers} <- collection_private(followers_data) do
-      info = %{
-        hide_follows: hide_follows,
-        follower_count: followers_count,
-        following_count: following_count,
-        hide_followers: hide_followers
-      }
-
-      info = Map.merge(user.info, info)
-      {:ok, Map.put(user, :info, info)}
+      {:ok,
+       %{
+         hide_follows: hide_follows,
+         follower_count: followers_count,
+         following_count: following_count,
+         hide_followers: hide_followers
+       }}
     else
       {:error, _} = e ->
         e
@@ -1043,8 +1087,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   defp maybe_update_follow_information(data) do
     with {:enabled, true} <-
            {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
-         {:ok, data} <- fetch_follow_information_for_user(data) do
-      data
+         {:ok, info} <- fetch_follow_information_for_user(data) do
+      info = Map.merge(data.info, info)
+      Map.put(data, :info, info)
     else
       {:enabled, false} ->
         data
@@ -1063,7 +1108,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
       {:ok, false}
     else
-      with {:ok, _data} <- Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
+      with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
+             Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
         {:ok, false}
       else
         {:error, {:ok, %{status: code}}} when code in [401, 403] ->