git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   6   alias Pleroma.Activity
   7   alias Pleroma.Activity.Ir.Topics
   8   alias Pleroma.Config
   9   alias Pleroma.Constants
  10   alias Pleroma.Conversation
  11   alias Pleroma.Conversation.Participation
  12   alias Pleroma.Notification
  13   alias Pleroma.Object
  14   alias Pleroma.Object.Containment
  15   alias Pleroma.Object.Fetcher
  16   alias Pleroma.Pagination
  17   alias Pleroma.Repo
  18   alias Pleroma.Upload
  19   alias Pleroma.User
  20   alias Pleroma.Web.ActivityPub.MRF
  21   alias Pleroma.Web.ActivityPub.Transmogrifier
  22   alias Pleroma.Web.ActivityPub.Utils
  23   alias Pleroma.Web.Streamer
  24   alias Pleroma.Web.WebFinger
  25   alias Pleroma.Workers.BackgroundWorker
  26
  27   import Ecto.Query
  28   import Pleroma.Web.ActivityPub.Utils
  29   import Pleroma.Web.ActivityPub.Visibility
  30
  31   require Logger
  32   require Pleroma.Constants
  33
  34   # For Announce activities, we filter the recipients based on following status for any actors
  35   # that match actual users.  See issue #164 for more information about why this is necessary.
  36   defp get_recipients(%{"type" => "Announce"} = data) do
  37     to = Map.get(data, "to", [])
  38     cc = Map.get(data, "cc", [])
  39     bcc = Map.get(data, "bcc", [])
  40     actor = User.get_cached_by_ap_id(data["actor"])
  41
  42     recipients =
  43       Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
  44         case User.get_cached_by_ap_id(recipient) do
  45           nil -> true
  46           user -> User.following?(user, actor)
  47         end
  48       end)
  49
  50     {recipients, to, cc}
  51   end
  52
  53   defp get_recipients(%{"type" => "Create"} = data) do
  54     to = Map.get(data, "to", [])
  55     cc = Map.get(data, "cc", [])
  56     bcc = Map.get(data, "bcc", [])
  57     actor = Map.get(data, "actor", [])
  58     recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
  59     {recipients, to, cc}
  60   end
  61
  62   defp get_recipients(data) do
  63     to = Map.get(data, "to", [])
  64     cc = Map.get(data, "cc", [])
  65     bcc = Map.get(data, "bcc", [])
  66     recipients = Enum.concat([to, cc, bcc])
  67     {recipients, to, cc}
  68   end
  69
  70   defp check_actor_is_active(actor) do
  71     if not is_nil(actor) do
  72       with user <- User.get_cached_by_ap_id(actor),
  73            false <- user.deactivated do
  74         true
  75       else
  76         _e -> false
  77       end
  78     else
  79       true
  80     end
  81   end
  82
  83   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
  84     limit = Config.get([:instance, :remote_limit])
  85     String.length(content) <= limit
  86   end
  87
  88   defp check_remote_limit(_), do: true
  89
  90   def increase_note_count_if_public(actor, object) do
  91     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  92   end
  93
  94   def decrease_note_count_if_public(actor, object) do
  95     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  96   end
  97
  98   def increase_replies_count_if_reply(%{
  99         "object" => %{"inReplyTo" => reply_ap_id} = object,
 100         "type" => "Create"
 101       }) do
 102     if is_public?(object) do
 103       Object.increase_replies_count(reply_ap_id)
 104     end
 105   end
 106
 107   def increase_replies_count_if_reply(_create_data), do: :noop
 108
 109   def decrease_replies_count_if_reply(%Object{
 110         data: %{"inReplyTo" => reply_ap_id} = object
 111       }) do
 112     if is_public?(object) do
 113       Object.decrease_replies_count(reply_ap_id)
 114     end
 115   end
 116
 117   def decrease_replies_count_if_reply(_object), do: :noop
 118
 119   def increase_poll_votes_if_vote(%{
 120         "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
 121         "type" => "Create",
 122         "actor" => actor
 123       }) do
 124     Object.increase_vote_count(reply_ap_id, name, actor)
 125   end
 126
 127   def increase_poll_votes_if_vote(_create_data), do: :noop
 128
 129   @spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
 130   def persist(object, meta) do
 131     with local <- Keyword.fetch!(meta, :local),
 132          {recipients, _, _} <- get_recipients(object),
 133          {:ok, activity} <-
 134            Repo.insert(%Activity{
 135              data: object,
 136              local: local,
 137              recipients: recipients,
 138              actor: object["actor"]
 139            }) do
 140       {:ok, activity, meta}
 141     end
 142   end
 143
 144   @spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 145   def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
 146     with nil <- Activity.normalize(map),
 147          map <- lazy_put_activity_defaults(map, fake),
 148          true <- bypass_actor_check || check_actor_is_active(map["actor"]),
 149          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
 150          {:ok, map} <- MRF.filter(map),
 151          {recipients, _, _} = get_recipients(map),
 152          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
 153          {:containment, :ok} <- {:containment, Containment.contain_child(map)},
 154          {:ok, map, object} <- insert_full_object(map) do
 155       {:ok, activity} =
 156         Repo.insert(%Activity{
 157           data: map,
 158           local: local,
 159           actor: map["actor"],
 160           recipients: recipients
 161         })
 162
 163       # Splice in the child object if we have one.
 164       activity =
 165         if not is_nil(object) do
 166           Map.put(activity, :object, object)
 167         else
 168           activity
 169         end
 170
 171       BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
 172
 173       Notification.create_notifications(activity)
 174
 175       conversation = create_or_bump_conversation(activity, map["actor"])
 176       participations = get_participations(conversation)
 177       stream_out(activity)
 178       stream_out_participations(participations)
 179       {:ok, activity}
 180     else
 181       %Activity{} = activity ->
 182         {:ok, activity}
 183
 184       {:fake, true, map, recipients} ->
 185         activity = %Activity{
 186           data: map,
 187           local: local,
 188           actor: map["actor"],
 189           recipients: recipients,
 190           id: "pleroma:fakeid"
 191         }
 192
 193         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 194         {:ok, activity}
 195
 196       error ->
 197         {:error, error}
 198     end
 199   end
 200
 201   defp create_or_bump_conversation(activity, actor) do
 202     with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
 203          %User{} = user <- User.get_cached_by_ap_id(actor),
 204          Participation.mark_as_read(user, conversation) do
 205       {:ok, conversation}
 206     end
 207   end
 208
 209   defp get_participations({:ok, conversation}) do
 210     conversation
 211     |> Repo.preload(:participations, force: true)
 212     |> Map.get(:participations)
 213   end
 214
 215   defp get_participations(_), do: []
 216
 217   def stream_out_participations(participations) do
 218     participations =
 219       participations
 220       |> Repo.preload(:user)
 221
 222     Streamer.stream("participation", participations)
 223   end
 224
 225   def stream_out_participations(%Object{data: %{"context" => context}}, user) do
 226     with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
 227          conversation = Repo.preload(conversation, :participations),
 228          last_activity_id =
 229            fetch_latest_activity_id_for_context(conversation.ap_id, %{
 230              "user" => user,
 231              "blocking_user" => user
 232            }) do
 233       if last_activity_id do
 234         stream_out_participations(conversation.participations)
 235       end
 236     end
 237   end
 238
 239   def stream_out_participations(_, _), do: :noop
 240
 241   def stream_out(%Activity{data: %{"type" => data_type}} = activity)
 242       when data_type in ["Create", "Announce", "Delete"] do
 243     activity
 244     |> Topics.get_activity_topics()
 245     |> Streamer.stream(activity)
 246   end
 247
 248   def stream_out(_activity) do
 249     :noop
 250   end
 251
 252   @spec create(map(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 253   def create(params, fake \\ false) do
 254     with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
 255       result
 256     end
 257   end
 258
 259   defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
 260     additional = params[:additional] || %{}
 261     # only accept false as false value
 262     local = !(params[:local] == false)
 263     published = params[:published]
 264     quick_insert? = Config.get([:env]) == :benchmark
 265
 266     with create_data <-
 267            make_create_data(
 268              %{to: to, actor: actor, published: published, context: context, object: object},
 269              additional
 270            ),
 271          {:ok, activity} <- insert(create_data, local, fake),
 272          {:fake, false, activity} <- {:fake, fake, activity},
 273          _ <- increase_replies_count_if_reply(create_data),
 274          _ <- increase_poll_votes_if_vote(create_data),
 275          {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
 276          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 277          :ok <- maybe_federate(activity) do
 278       {:ok, activity}
 279     else
 280       {:quick_insert, true, activity} ->
 281         {:ok, activity}
 282
 283       {:fake, true, activity} ->
 284         {:ok, activity}
 285
 286       {:error, message} ->
 287         Repo.rollback(message)
 288     end
 289   end
 290
 291   @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
 292   def listen(%{to: to, actor: actor, context: context, object: object} = params) do
 293     additional = params[:additional] || %{}
 294     # only accept false as false value
 295     local = !(params[:local] == false)
 296     published = params[:published]
 297
 298     with listen_data <-
 299            make_listen_data(
 300              %{to: to, actor: actor, published: published, context: context, object: object},
 301              additional
 302            ),
 303          {:ok, activity} <- insert(listen_data, local),
 304          :ok <- maybe_federate(activity) do
 305       {:ok, activity}
 306     end
 307   end
 308
 309   @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
 310   def accept(params) do
 311     accept_or_reject("Accept", params)
 312   end
 313
 314   @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
 315   def reject(params) do
 316     accept_or_reject("Reject", params)
 317   end
 318
 319   @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
 320   def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
 321     local = Map.get(params, :local, true)
 322     activity_id = Map.get(params, :activity_id, nil)
 323
 324     with data <-
 325            %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
 326            |> Utils.maybe_put("id", activity_id),
 327          {:ok, activity} <- insert(data, local),
 328          :ok <- maybe_federate(activity) do
 329       {:ok, activity}
 330     end
 331   end
 332
 333   @spec update(map()) :: {:ok, Activity.t()} | {:error, any()}
 334   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 335     local = !(params[:local] == false)
 336     activity_id = params[:activity_id]
 337
 338     with data <- %{
 339            "to" => to,
 340            "cc" => cc,
 341            "type" => "Update",
 342            "actor" => actor,
 343            "object" => object
 344          },
 345          data <- Utils.maybe_put(data, "id", activity_id),
 346          {:ok, activity} <- insert(data, local),
 347          :ok <- maybe_federate(activity) do
 348       {:ok, activity}
 349     end
 350   end
 351
 352   @spec react_with_emoji(User.t(), Object.t(), String.t(), keyword()) ::
 353           {:ok, Activity.t(), Object.t()} | {:error, any()}
 354   def react_with_emoji(user, object, emoji, options \\ []) do
 355     with {:ok, result} <-
 356            Repo.transaction(fn -> do_react_with_emoji(user, object, emoji, options) end) do
 357       result
 358     end
 359   end
 360
 361   defp do_react_with_emoji(user, object, emoji, options) do
 362     with local <- Keyword.get(options, :local, true),
 363          activity_id <- Keyword.get(options, :activity_id, nil),
 364          true <- Pleroma.Emoji.is_unicode_emoji?(emoji),
 365          reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
 366          {:ok, activity} <- insert(reaction_data, local),
 367          {:ok, object} <- add_emoji_reaction_to_object(activity, object),
 368          :ok <- maybe_federate(activity) do
 369       {:ok, activity, object}
 370     else
 371       false -> {:error, false}
 372       {:error, error} -> Repo.rollback(error)
 373     end
 374   end
 375
 376   @spec unreact_with_emoji(User.t(), String.t(), keyword()) ::
 377           {:ok, Activity.t(), Object.t()} | {:error, any()}
 378   def unreact_with_emoji(user, reaction_id, options \\ []) do
 379     with {:ok, result} <-
 380            Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
 381       result
 382     end
 383   end
 384
 385   defp do_unreact_with_emoji(user, reaction_id, options) do
 386     with local <- Keyword.get(options, :local, true),
 387          activity_id <- Keyword.get(options, :activity_id, nil),
 388          user_ap_id <- user.ap_id,
 389          %Activity{actor: ^user_ap_id} = reaction_activity <- Activity.get_by_ap_id(reaction_id),
 390          object <- Object.normalize(reaction_activity),
 391          unreact_data <- make_undo_data(user, reaction_activity, activity_id),
 392          {:ok, activity} <- insert(unreact_data, local),
 393          {:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
 394          :ok <- maybe_federate(activity) do
 395       {:ok, activity, object}
 396     else
 397       {:error, error} -> Repo.rollback(error)
 398     end
 399   end
 400
 401   @spec unlike(User.t(), Object.t(), String.t() | nil, boolean()) ::
 402           {:ok, Activity.t(), Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
 403   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
 404     with {:ok, result} <-
 405            Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
 406       result
 407     end
 408   end
 409
 410   defp do_unlike(actor, object, activity_id, local) do
 411     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 412          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 413          {:ok, unlike_activity} <- insert(unlike_data, local),
 414          {:ok, _activity} <- Repo.delete(like_activity),
 415          {:ok, object} <- remove_like_from_object(like_activity, object),
 416          :ok <- maybe_federate(unlike_activity) do
 417       {:ok, unlike_activity, like_activity, object}
 418     else
 419       nil -> {:ok, object}
 420       {:error, error} -> Repo.rollback(error)
 421     end
 422   end
 423
 424   @spec announce(User.t(), Object.t(), String.t() | nil, boolean(), boolean()) ::
 425           {:ok, Activity.t(), Object.t()} | {:error, any()}
 426   def announce(
 427         %User{ap_id: _} = user,
 428         %Object{data: %{"id" => _}} = object,
 429         activity_id \\ nil,
 430         local \\ true,
 431         public \\ true
 432       ) do
 433     with {:ok, result} <-
 434            Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
 435       result
 436     end
 437   end
 438
 439   defp do_announce(user, object, activity_id, local, public) do
 440     with true <- is_announceable?(object, user, public),
 441          announce_data <- make_announce_data(user, object, activity_id, public),
 442          {:ok, activity} <- insert(announce_data, local),
 443          {:ok, object} <- add_announce_to_object(activity, object),
 444          :ok <- maybe_federate(activity) do
 445       {:ok, activity, object}
 446     else
 447       false -> {:error, false}
 448       {:error, error} -> Repo.rollback(error)
 449     end
 450   end
 451
 452   @spec unannounce(User.t(), Object.t(), String.t() | nil, boolean()) ::
 453           {:ok, Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
 454   def unannounce(
 455         %User{} = actor,
 456         %Object{} = object,
 457         activity_id \\ nil,
 458         local \\ true
 459       ) do
 460     with {:ok, result} <-
 461            Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
 462       result
 463     end
 464   end
 465
 466   defp do_unannounce(actor, object, activity_id, local) do
 467     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 468          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 469          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 470          :ok <- maybe_federate(unannounce_activity),
 471          {:ok, _activity} <- Repo.delete(announce_activity),
 472          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 473       {:ok, unannounce_activity, object}
 474     else
 475       nil -> {:ok, object}
 476       {:error, error} -> Repo.rollback(error)
 477     end
 478   end
 479
 480   @spec follow(User.t(), User.t(), String.t() | nil, boolean()) ::
 481           {:ok, Activity.t()} | {:error, any()}
 482   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 483     with {:ok, result} <-
 484            Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
 485       result
 486     end
 487   end
 488
 489   defp do_follow(follower, followed, activity_id, local) do
 490     with data <- make_follow_data(follower, followed, activity_id),
 491          {:ok, activity} <- insert(data, local),
 492          :ok <- maybe_federate(activity) do
 493       {:ok, activity}
 494     else
 495       {:error, error} -> Repo.rollback(error)
 496     end
 497   end
 498
 499   @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
 500           {:ok, Activity.t()} | nil | {:error, any()}
 501   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 502     with {:ok, result} <-
 503            Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
 504       result
 505     end
 506   end
 507
 508   defp do_unfollow(follower, followed, activity_id, local) do
 509     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 510          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 511          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 512          {:ok, activity} <- insert(unfollow_data, local),
 513          :ok <- maybe_federate(activity) do
 514       {:ok, activity}
 515     else
 516       nil -> nil
 517       {:error, error} -> Repo.rollback(error)
 518     end
 519   end
 520
 521   @spec delete(User.t() | Object.t(), keyword()) :: {:ok, User.t() | Object.t()} | {:error, any()}
 522   def delete(entity, options \\ []) do
 523     with {:ok, result} <- Repo.transaction(fn -> do_delete(entity, options) end) do
 524       result
 525     end
 526   end
 527
 528   defp do_delete(%User{ap_id: ap_id, follower_address: follower_address} = user, _) do
 529     with data <- %{
 530            "to" => [follower_address],
 531            "type" => "Delete",
 532            "actor" => ap_id,
 533            "object" => %{"type" => "Person", "id" => ap_id}
 534          },
 535          {:ok, activity} <- insert(data, true, true, true),
 536          :ok <- maybe_federate(activity) do
 537       {:ok, user}
 538     end
 539   end
 540
 541   defp do_delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options) do
 542     local = Keyword.get(options, :local, true)
 543     activity_id = Keyword.get(options, :activity_id, nil)
 544     actor = Keyword.get(options, :actor, actor)
 545
 546     user = User.get_cached_by_ap_id(actor)
 547     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 548
 549     with create_activity <- Activity.get_create_by_object_ap_id(id),
 550          data <-
 551            %{
 552              "type" => "Delete",
 553              "actor" => actor,
 554              "object" => id,
 555              "to" => to,
 556              "deleted_activity_id" => create_activity && create_activity.id
 557            }
 558            |> maybe_put("id", activity_id),
 559          {:ok, activity} <- insert(data, local, false),
 560          {:ok, object, _create_activity} <- Object.delete(object),
 561          stream_out_participations(object, user),
 562          _ <- decrease_replies_count_if_reply(object),
 563          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 564          :ok <- maybe_federate(activity) do
 565       {:ok, activity}
 566     else
 567       {:error, error} ->
 568         Repo.rollback(error)
 569     end
 570   end
 571
 572   defp do_delete(%Object{data: %{"type" => "Tombstone", "id" => ap_id}}, _) do
 573     activity =
 574       ap_id
 575       |> Activity.Queries.by_object_id()
 576       |> Activity.Queries.by_type("Delete")
 577       |> Repo.one()
 578
 579     {:ok, activity}
 580   end
 581
 582   @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
 583           {:ok, Activity.t()} | {:error, any()}
 584   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 585     with {:ok, result} <-
 586            Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
 587       result
 588     end
 589   end
 590
 591   defp do_block(blocker, blocked, activity_id, local) do
 592     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
 593     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 594
 595     if unfollow_blocked do
 596       follow_activity = fetch_latest_follow(blocker, blocked)
 597       if follow_activity, do: unfollow(blocker, blocked, nil, local)
 598     end
 599
 600     with true <- outgoing_blocks,
 601          block_data <- make_block_data(blocker, blocked, activity_id),
 602          {:ok, activity} <- insert(block_data, local),
 603          :ok <- maybe_federate(activity) do
 604       {:ok, activity}
 605     else
 606       {:error, error} -> Repo.rollback(error)
 607     end
 608   end
 609
 610   @spec unblock(User.t(), User.t(), String.t() | nil, boolean()) ::
 611           {:ok, Activity.t()} | {:error, any()} | nil
 612   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 613     with {:ok, result} <-
 614            Repo.transaction(fn -> do_unblock(blocker, blocked, activity_id, local) end) do
 615       result
 616     end
 617   end
 618
 619   defp do_unblock(blocker, blocked, activity_id, local) do
 620     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 621          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 622          {:ok, activity} <- insert(unblock_data, local),
 623          :ok <- maybe_federate(activity) do
 624       {:ok, activity}
 625     else
 626       nil -> nil
 627       {:error, error} -> Repo.rollback(error)
 628     end
 629   end
 630
 631   @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
 632   def flag(
 633         %{
 634           actor: actor,
 635           context: _context,
 636           account: account,
 637           statuses: statuses,
 638           content: content
 639         } = params
 640       ) do
 641     # only accept false as false value
 642     local = !(params[:local] == false)
 643     forward = !(params[:forward] == false)
 644
 645     additional = params[:additional] || %{}
 646
 647     additional =
 648       if forward do
 649         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 650       else
 651         Map.merge(additional, %{"to" => [], "cc" => []})
 652       end
 653
 654     with flag_data <- make_flag_data(params, additional),
 655          {:ok, activity} <- insert(flag_data, local),
 656          {:ok, stripped_activity} <- strip_report_status_data(activity),
 657          :ok <- maybe_federate(stripped_activity) do
 658       User.all_superusers()
 659       |> Enum.filter(fn user -> not is_nil(user.email) end)
 660       |> Enum.each(fn superuser ->
 661         superuser
 662         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 663         |> Pleroma.Emails.Mailer.deliver_async()
 664       end)
 665
 666       {:ok, activity}
 667     end
 668   end
 669
 670   @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 671   def move(%User{} = origin, %User{} = target, local \\ true) do
 672     params = %{
 673       "type" => "Move",
 674       "actor" => origin.ap_id,
 675       "object" => origin.ap_id,
 676       "target" => target.ap_id
 677     }
 678
 679     with true <- origin.ap_id in target.also_known_as,
 680          {:ok, activity} <- insert(params, local) do
 681       maybe_federate(activity)
 682
 683       BackgroundWorker.enqueue("move_following", %{
 684         "origin_id" => origin.id,
 685         "target_id" => target.id
 686       })
 687
 688       {:ok, activity}
 689     else
 690       false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
 691       err -> err
 692     end
 693   end
 694
 695   def fetch_activities_for_context_query(context, opts) do
 696     public = [Constants.as_public()]
 697
 698     recipients =
 699       if opts["user"],
 700         do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
 701         else: public
 702
 703     from(activity in Activity)
 704     |> maybe_preload_objects(opts)
 705     |> maybe_preload_bookmarks(opts)
 706     |> maybe_set_thread_muted_field(opts)
 707     |> restrict_blocked(opts)
 708     |> restrict_recipients(recipients, opts["user"])
 709     |> where(
 710       [activity],
 711       fragment(
 712         "?->>'type' = ? and ?->>'context' = ?",
 713         activity.data,
 714         "Create",
 715         activity.data,
 716         ^context
 717       )
 718     )
 719     |> exclude_poll_votes(opts)
 720     |> exclude_id(opts)
 721     |> order_by([activity], desc: activity.id)
 722   end
 723
 724   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 725   def fetch_activities_for_context(context, opts \\ %{}) do
 726     context
 727     |> fetch_activities_for_context_query(opts)
 728     |> Repo.all()
 729   end
 730
 731   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 732           FlakeId.Ecto.CompatType.t() | nil
 733   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 734     context
 735     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 736     |> limit(1)
 737     |> select([a], a.id)
 738     |> Repo.one()
 739   end
 740
 741   @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
 742   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 743     opts = Map.drop(opts, ["user"])
 744
 745     [Constants.as_public()]
 746     |> fetch_activities_query(opts)
 747     |> restrict_unlisted()
 748     |> Pagination.fetch_paginated(opts, pagination)
 749   end
 750
 751   @valid_visibilities ~w[direct unlisted public private]
 752
 753   defp restrict_visibility(query, %{visibility: visibility})
 754        when is_list(visibility) do
 755     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 756       query =
 757         from(
 758           a in query,
 759           where:
 760             fragment(
 761               "activity_visibility(?, ?, ?) = ANY (?)",
 762               a.actor,
 763               a.recipients,
 764               a.data,
 765               ^visibility
 766             )
 767         )
 768
 769       query
 770     else
 771       Logger.error("Could not restrict visibility to #{visibility}")
 772     end
 773   end
 774
 775   defp restrict_visibility(query, %{visibility: visibility})
 776        when visibility in @valid_visibilities do
 777     from(
 778       a in query,
 779       where:
 780         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 781     )
 782   end
 783
 784   defp restrict_visibility(_query, %{visibility: visibility})
 785        when visibility not in @valid_visibilities do
 786     Logger.error("Could not restrict visibility to #{visibility}")
 787   end
 788
 789   defp restrict_visibility(query, _visibility), do: query
 790
 791   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 792        when is_list(visibility) do
 793     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 794       from(
 795         a in query,
 796         where:
 797           not fragment(
 798             "activity_visibility(?, ?, ?) = ANY (?)",
 799             a.actor,
 800             a.recipients,
 801             a.data,
 802             ^visibility
 803           )
 804       )
 805     else
 806       Logger.error("Could not exclude visibility to #{visibility}")
 807       query
 808     end
 809   end
 810
 811   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 812        when visibility in @valid_visibilities do
 813     from(
 814       a in query,
 815       where:
 816         not fragment(
 817           "activity_visibility(?, ?, ?) = ?",
 818           a.actor,
 819           a.recipients,
 820           a.data,
 821           ^visibility
 822         )
 823     )
 824   end
 825
 826   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 827        when visibility not in @valid_visibilities do
 828     Logger.error("Could not exclude visibility to #{visibility}")
 829     query
 830   end
 831
 832   defp exclude_visibility(query, _visibility), do: query
 833
 834   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 835     do: query
 836
 837   defp restrict_thread_visibility(
 838          query,
 839          %{"user" => %User{skip_thread_containment: true}},
 840          _
 841        ),
 842        do: query
 843
 844   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 845     from(
 846       a in query,
 847       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 848     )
 849   end
 850
 851   defp restrict_thread_visibility(query, _, _), do: query
 852
 853   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 854     params =
 855       params
 856       |> Map.put("user", reading_user)
 857       |> Map.put("actor_id", user.ap_id)
 858
 859     recipients =
 860       user_activities_recipients(%{
 861         "godmode" => params["godmode"],
 862         "reading_user" => reading_user
 863       })
 864
 865     fetch_activities(recipients, params)
 866     |> Enum.reverse()
 867   end
 868
 869   def fetch_user_activities(user, reading_user, params \\ %{}) do
 870     params =
 871       params
 872       |> Map.put("type", ["Create", "Announce"])
 873       |> Map.put("user", reading_user)
 874       |> Map.put("actor_id", user.ap_id)
 875       |> Map.put("pinned_activity_ids", user.pinned_activities)
 876
 877     params =
 878       if User.blocks?(reading_user, user) do
 879         params
 880       else
 881         params
 882         |> Map.put("blocking_user", reading_user)
 883         |> Map.put("muting_user", reading_user)
 884       end
 885
 886     recipients =
 887       user_activities_recipients(%{
 888         "godmode" => params["godmode"],
 889         "reading_user" => reading_user
 890       })
 891
 892     fetch_activities(recipients, params)
 893     |> Enum.reverse()
 894   end
 895
 896   def fetch_statuses(reading_user, params) do
 897     params =
 898       params
 899       |> Map.put("type", ["Create", "Announce"])
 900
 901     recipients =
 902       user_activities_recipients(%{
 903         "godmode" => params["godmode"],
 904         "reading_user" => reading_user
 905       })
 906
 907     fetch_activities(recipients, params, :offset)
 908     |> Enum.reverse()
 909   end
 910
 911   defp user_activities_recipients(%{"godmode" => true}) do
 912     []
 913   end
 914
 915   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 916     if reading_user do
 917       [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 918     else
 919       [Constants.as_public()]
 920     end
 921   end
 922
 923   defp restrict_since(query, %{"since_id" => ""}), do: query
 924
 925   defp restrict_since(query, %{"since_id" => since_id}) do
 926     from(activity in query, where: activity.id > ^since_id)
 927   end
 928
 929   defp restrict_since(query, _), do: query
 930
 931   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 932     raise "Can't use the child object without preloading!"
 933   end
 934
 935   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 936        when is_list(tag_reject) and tag_reject != [] do
 937     from(
 938       [_activity, object] in query,
 939       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 940     )
 941   end
 942
 943   defp restrict_tag_reject(query, _), do: query
 944
 945   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 946     raise "Can't use the child object without preloading!"
 947   end
 948
 949   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 950        when is_list(tag_all) and tag_all != [] do
 951     from(
 952       [_activity, object] in query,
 953       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 954     )
 955   end
 956
 957   defp restrict_tag_all(query, _), do: query
 958
 959   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 960     raise "Can't use the child object without preloading!"
 961   end
 962
 963   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 964     from(
 965       [_activity, object] in query,
 966       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 967     )
 968   end
 969
 970   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 971     from(
 972       [_activity, object] in query,
 973       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 974     )
 975   end
 976
 977   defp restrict_tag(query, _), do: query
 978
 979   defp restrict_recipients(query, [], _user), do: query
 980
 981   defp restrict_recipients(query, recipients, nil) do
 982     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 983   end
 984
 985   defp restrict_recipients(query, recipients, user) do
 986     from(
 987       activity in query,
 988       where: fragment("? && ?", ^recipients, activity.recipients),
 989       or_where: activity.actor == ^user.ap_id
 990     )
 991   end
 992
 993   defp restrict_local(query, %{"local_only" => true}) do
 994     from(activity in query, where: activity.local == true)
 995   end
 996
 997   defp restrict_local(query, _), do: query
 998
 999   defp restrict_actor(query, %{"actor_id" => actor_id}) do
1000     from(activity in query, where: activity.actor == ^actor_id)
1001   end
1002
1003   defp restrict_actor(query, _), do: query
1004
1005   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
1006     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
1007   end
1008
1009   defp restrict_type(query, %{"type" => type}) do
1010     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
1011   end
1012
1013   defp restrict_type(query, _), do: query
1014
1015   defp restrict_state(query, %{"state" => state}) do
1016     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
1017   end
1018
1019   defp restrict_state(query, _), do: query
1020
1021   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
1022     from(
1023       [_activity, object] in query,
1024       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
1025     )
1026   end
1027
1028   defp restrict_favorited_by(query, _), do: query
1029
1030   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
1031     raise "Can't use the child object without preloading!"
1032   end
1033
1034   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
1035     from(
1036       [_activity, object] in query,
1037       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
1038     )
1039   end
1040
1041   defp restrict_media(query, _), do: query
1042
1043   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
1044     from(
1045       [_activity, object] in query,
1046       where: fragment("?->>'inReplyTo' is null", object.data)
1047     )
1048   end
1049
1050   defp restrict_replies(query, _), do: query
1051
1052   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
1053     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
1054   end
1055
1056   defp restrict_reblogs(query, _), do: query
1057
1058   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
1059
1060   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
1061     mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
1062
1063     query =
1064       from([activity] in query,
1065         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
1066         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
1067       )
1068
1069     unless opts["skip_preload"] do
1070       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
1071     else
1072       query
1073     end
1074   end
1075
1076   defp restrict_muted(query, _), do: query
1077
1078   defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
1079     blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
1080     domain_blocks = user.domain_blocks || []
1081
1082     following_ap_ids = User.get_friends_ap_ids(user)
1083
1084     query =
1085       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
1086
1087     from(
1088       [activity, object: o] in query,
1089       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
1090       where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
1091       where:
1092         fragment(
1093           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1094           activity.data,
1095           activity.data,
1096           ^blocked_ap_ids
1097         ),
1098       where:
1099         fragment(
1100           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
1101           activity.actor,
1102           ^domain_blocks,
1103           activity.actor,
1104           ^following_ap_ids
1105         ),
1106       where:
1107         fragment(
1108           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
1109           o.data,
1110           ^domain_blocks,
1111           o.data,
1112           ^following_ap_ids
1113         )
1114     )
1115   end
1116
1117   defp restrict_blocked(query, _), do: query
1118
1119   defp restrict_unlisted(query) do
1120     from(
1121       activity in query,
1122       where:
1123         fragment(
1124           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
1125           activity.data,
1126           ^[Constants.as_public()]
1127         )
1128     )
1129   end
1130
1131   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
1132     from(activity in query, where: activity.id in ^ids)
1133   end
1134
1135   defp restrict_pinned(query, _), do: query
1136
1137   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1138     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1139
1140     from(
1141       activity in query,
1142       where:
1143         fragment(
1144           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1145           activity.data,
1146           activity.actor,
1147           ^muted_reblogs
1148         )
1149     )
1150   end
1151
1152   defp restrict_muted_reblogs(query, _), do: query
1153
1154   defp restrict_instance(query, %{"instance" => instance}) do
1155     users =
1156       from(
1157         u in User,
1158         select: u.ap_id,
1159         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1160       )
1161       |> Repo.all()
1162
1163     from(activity in query, where: activity.actor in ^users)
1164   end
1165
1166   defp restrict_instance(query, _), do: query
1167
1168   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1169
1170   defp exclude_poll_votes(query, _) do
1171     if has_named_binding?(query, :object) do
1172       from([activity, object: o] in query,
1173         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1174       )
1175     else
1176       query
1177     end
1178   end
1179
1180   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1181     from(activity in query, where: activity.id != ^id)
1182   end
1183
1184   defp exclude_id(query, _), do: query
1185
1186   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1187
1188   defp maybe_preload_objects(query, _) do
1189     query
1190     |> Activity.with_preloaded_object()
1191   end
1192
1193   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1194
1195   defp maybe_preload_bookmarks(query, opts) do
1196     query
1197     |> Activity.with_preloaded_bookmark(opts["user"])
1198   end
1199
1200   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1201     query
1202     |> Activity.with_preloaded_report_notes()
1203   end
1204
1205   defp maybe_preload_report_notes(query, _), do: query
1206
1207   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1208
1209   defp maybe_set_thread_muted_field(query, opts) do
1210     query
1211     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1212   end
1213
1214   defp maybe_order(query, %{order: :desc}) do
1215     query
1216     |> order_by(desc: :id)
1217   end
1218
1219   defp maybe_order(query, %{order: :asc}) do
1220     query
1221     |> order_by(asc: :id)
1222   end
1223
1224   defp maybe_order(query, _), do: query
1225
1226   defp fetch_activities_query_ap_ids_ops(opts) do
1227     source_user = opts["muting_user"]
1228     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1229
1230     ap_id_relationships =
1231       ap_id_relationships ++
1232         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1233           [:block]
1234         else
1235           []
1236         end
1237
1238     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1239
1240     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1241     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1242
1243     restrict_muted_reblogs_opts =
1244       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1245
1246     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1247   end
1248
1249   def fetch_activities_query(recipients, opts \\ %{}) do
1250     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1251       fetch_activities_query_ap_ids_ops(opts)
1252
1253     config = %{
1254       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1255     }
1256
1257     Activity
1258     |> maybe_preload_objects(opts)
1259     |> maybe_preload_bookmarks(opts)
1260     |> maybe_preload_report_notes(opts)
1261     |> maybe_set_thread_muted_field(opts)
1262     |> maybe_order(opts)
1263     |> restrict_recipients(recipients, opts["user"])
1264     |> restrict_tag(opts)
1265     |> restrict_tag_reject(opts)
1266     |> restrict_tag_all(opts)
1267     |> restrict_since(opts)
1268     |> restrict_local(opts)
1269     |> restrict_actor(opts)
1270     |> restrict_type(opts)
1271     |> restrict_state(opts)
1272     |> restrict_favorited_by(opts)
1273     |> restrict_blocked(restrict_blocked_opts)
1274     |> restrict_muted(restrict_muted_opts)
1275     |> restrict_media(opts)
1276     |> restrict_visibility(opts)
1277     |> restrict_thread_visibility(opts, config)
1278     |> restrict_replies(opts)
1279     |> restrict_reblogs(opts)
1280     |> restrict_pinned(opts)
1281     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1282     |> restrict_instance(opts)
1283     |> Activity.restrict_deactivated_users()
1284     |> exclude_poll_votes(opts)
1285     |> exclude_visibility(opts)
1286   end
1287
1288   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1289     list_memberships = Pleroma.List.memberships(opts["user"])
1290
1291     fetch_activities_query(recipients ++ list_memberships, opts)
1292     |> Pagination.fetch_paginated(opts, pagination)
1293     |> Enum.reverse()
1294     |> maybe_update_cc(list_memberships, opts["user"])
1295   end
1296
1297   @doc """
1298   Fetch favorites activities of user with order by sort adds to favorites
1299   """
1300   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1301   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1302     user.ap_id
1303     |> Activity.Queries.by_actor()
1304     |> Activity.Queries.by_type("Like")
1305     |> Activity.with_joined_object()
1306     |> Object.with_joined_activity()
1307     |> select([_like, object, activity], %{activity | object: object})
1308     |> order_by([like, _, _], desc: like.id)
1309     |> Pagination.fetch_paginated(
1310       Map.merge(params, %{"skip_order" => true}),
1311       pagination,
1312       :object_activity
1313     )
1314   end
1315
1316   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1317        when is_list(list_memberships) and length(list_memberships) > 0 do
1318     Enum.map(activities, fn
1319       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1320         if Enum.any?(bcc, &(&1 in list_memberships)) do
1321           update_in(activity.data["cc"], &[user_ap_id | &1])
1322         else
1323           activity
1324         end
1325
1326       activity ->
1327         activity
1328     end)
1329   end
1330
1331   defp maybe_update_cc(activities, _, _), do: activities
1332
1333   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1334     from(activity in query,
1335       where:
1336         fragment("? && ?", activity.recipients, ^recipients) or
1337           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1338              ^Constants.as_public() in activity.recipients)
1339     )
1340   end
1341
1342   def fetch_activities_bounded(
1343         recipients,
1344         recipients_with_public,
1345         opts \\ %{},
1346         pagination \\ :keyset
1347       ) do
1348     fetch_activities_query([], opts)
1349     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1350     |> Pagination.fetch_paginated(opts, pagination)
1351     |> Enum.reverse()
1352   end
1353
1354   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1355   def upload(file, opts \\ []) do
1356     with {:ok, data} <- Upload.store(file, opts) do
1357       obj_data =
1358         if opts[:actor] do
1359           Map.put(data, "actor", opts[:actor])
1360         else
1361           data
1362         end
1363
1364       Repo.insert(%Object{data: obj_data})
1365     end
1366   end
1367
1368   @spec get_actor_url(any()) :: binary() | nil
1369   defp get_actor_url(url) when is_binary(url), do: url
1370   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1371
1372   defp get_actor_url(url) when is_list(url) do
1373     url
1374     |> List.first()
1375     |> get_actor_url()
1376   end
1377
1378   defp get_actor_url(_url), do: nil
1379
1380   defp object_to_user_data(data) do
1381     avatar =
1382       data["icon"]["url"] &&
1383         %{
1384           "type" => "Image",
1385           "url" => [%{"href" => data["icon"]["url"]}]
1386         }
1387
1388     banner =
1389       data["image"]["url"] &&
1390         %{
1391           "type" => "Image",
1392           "url" => [%{"href" => data["image"]["url"]}]
1393         }
1394
1395     fields =
1396       data
1397       |> Map.get("attachment", [])
1398       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1399       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1400
1401     emojis =
1402       data
1403       |> Map.get("tag", [])
1404       |> Enum.filter(fn
1405         %{"type" => "Emoji"} -> true
1406         _ -> false
1407       end)
1408       |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1409         Map.put(acc, String.trim(name, ":"), url)
1410       end)
1411
1412     locked = data["manuallyApprovesFollowers"] || false
1413     data = Transmogrifier.maybe_fix_user_object(data)
1414     discoverable = data["discoverable"] || false
1415     invisible = data["invisible"] || false
1416     actor_type = data["type"] || "Person"
1417
1418     public_key =
1419       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1420         data["publicKey"]["publicKeyPem"]
1421       else
1422         nil
1423       end
1424
1425     shared_inbox =
1426       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1427         data["endpoints"]["sharedInbox"]
1428       else
1429         nil
1430       end
1431
1432     user_data = %{
1433       ap_id: data["id"],
1434       uri: get_actor_url(data["url"]),
1435       ap_enabled: true,
1436       banner: banner,
1437       fields: fields,
1438       emoji: emojis,
1439       locked: locked,
1440       discoverable: discoverable,
1441       invisible: invisible,
1442       avatar: avatar,
1443       name: data["name"],
1444       follower_address: data["followers"],
1445       following_address: data["following"],
1446       bio: data["summary"],
1447       actor_type: actor_type,
1448       also_known_as: Map.get(data, "alsoKnownAs", []),
1449       public_key: public_key,
1450       inbox: data["inbox"],
1451       shared_inbox: shared_inbox
1452     }
1453
1454     # nickname can be nil because of virtual actors
1455     user_data =
1456       if data["preferredUsername"] do
1457         Map.put(
1458           user_data,
1459           :nickname,
1460           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1461         )
1462       else
1463         Map.put(user_data, :nickname, nil)
1464       end
1465
1466     {:ok, user_data}
1467   end
1468
1469   def fetch_follow_information_for_user(user) do
1470     with {:ok, following_data} <-
1471            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1472          {:ok, hide_follows} <- collection_private(following_data),
1473          {:ok, followers_data} <-
1474            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1475          {:ok, hide_followers} <- collection_private(followers_data) do
1476       {:ok,
1477        %{
1478          hide_follows: hide_follows,
1479          follower_count: normalize_counter(followers_data["totalItems"]),
1480          following_count: normalize_counter(following_data["totalItems"]),
1481          hide_followers: hide_followers
1482        }}
1483     else
1484       {:error, _} = e -> e
1485       e -> {:error, e}
1486     end
1487   end
1488
1489   defp normalize_counter(counter) when is_integer(counter), do: counter
1490   defp normalize_counter(_), do: 0
1491
1492   defp maybe_update_follow_information(data) do
1493     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1494          {:ok, info} <- fetch_follow_information_for_user(data) do
1495       info = Map.merge(data[:info] || %{}, info)
1496       Map.put(data, :info, info)
1497     else
1498       {:enabled, false} ->
1499         data
1500
1501       e ->
1502         Logger.error(
1503           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1504         )
1505
1506         data
1507     end
1508   end
1509
1510   defp collection_private(%{"first" => %{"type" => type}})
1511        when type in ["CollectionPage", "OrderedCollectionPage"],
1512        do: {:ok, false}
1513
1514   defp collection_private(%{"first" => first}) do
1515     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1516            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1517       {:ok, false}
1518     else
1519       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1520       {:error, _} = e -> e
1521       e -> {:error, e}
1522     end
1523   end
1524
1525   defp collection_private(_data), do: {:ok, true}
1526
1527   def user_data_from_user_object(data) do
1528     with {:ok, data} <- MRF.filter(data),
1529          {:ok, data} <- object_to_user_data(data) do
1530       {:ok, data}
1531     else
1532       e -> {:error, e}
1533     end
1534   end
1535
1536   def fetch_and_prepare_user_from_ap_id(ap_id) do
1537     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1538          {:ok, data} <- user_data_from_user_object(data),
1539          data <- maybe_update_follow_information(data) do
1540       {:ok, data}
1541     else
1542       {:error, "Object has been deleted"} = e ->
1543         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1544         {:error, e}
1545
1546       e ->
1547         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1548         {:error, e}
1549     end
1550   end
1551
1552   def make_user_from_ap_id(ap_id) do
1553     user = User.get_cached_by_ap_id(ap_id)
1554
1555     if user && !User.ap_enabled?(user) do
1556       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1557     else
1558       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1559         if user do
1560           user
1561           |> User.remote_user_changeset(data)
1562           |> User.update_and_set_cache()
1563         else
1564           data
1565           |> User.remote_user_changeset()
1566           |> Repo.insert()
1567           |> User.set_cache()
1568         end
1569       else
1570         e -> {:error, e}
1571       end
1572     end
1573   end
1574
1575   def make_user_from_nickname(nickname) do
1576     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1577       make_user_from_ap_id(ap_id)
1578     else
1579       _e -> {:error, "No AP id in WebFinger"}
1580     end
1581   end
1582
1583   # filter out broken threads
1584   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1585     entire_thread_visible_for_user?(activity, user)
1586   end
1587
1588   # do post-processing on a specific activity
1589   def contain_activity(%Activity{} = activity, %User{} = user) do
1590     contain_broken_threads(activity, user)
1591   end
1592
1593   def fetch_direct_messages_query do
1594     Activity
1595     |> restrict_type(%{"type" => "Create"})
1596     |> restrict_visibility(%{visibility: "direct"})
1597     |> order_by([activity], asc: activity.id)
1598   end
1599 end