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 unreact_with_emoji(User.t(), String.t(), keyword()) ::
 353           {:ok, Activity.t(), Object.t()} | {:error, any()}
 354   def unreact_with_emoji(user, reaction_id, options \\ []) do
 355     with {:ok, result} <-
 356            Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
 357       result
 358     end
 359   end
 360
 361   defp do_unreact_with_emoji(user, reaction_id, options) do
 362     with local <- Keyword.get(options, :local, true),
 363          activity_id <- Keyword.get(options, :activity_id, nil),
 364          user_ap_id <- user.ap_id,
 365          %Activity{actor: ^user_ap_id} = reaction_activity <- Activity.get_by_ap_id(reaction_id),
 366          object <- Object.normalize(reaction_activity),
 367          unreact_data <- make_undo_data(user, reaction_activity, activity_id),
 368          {:ok, activity} <- insert(unreact_data, local),
 369          {:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
 370          :ok <- maybe_federate(activity) do
 371       {:ok, activity, object}
 372     else
 373       {:error, error} -> Repo.rollback(error)
 374     end
 375   end
 376
 377   @spec unlike(User.t(), Object.t(), String.t() | nil, boolean()) ::
 378           {:ok, Activity.t(), Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
 379   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
 380     with {:ok, result} <-
 381            Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
 382       result
 383     end
 384   end
 385
 386   defp do_unlike(actor, object, activity_id, local) do
 387     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 388          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 389          {:ok, unlike_activity} <- insert(unlike_data, local),
 390          {:ok, _activity} <- Repo.delete(like_activity),
 391          {:ok, object} <- remove_like_from_object(like_activity, object),
 392          :ok <- maybe_federate(unlike_activity) do
 393       {:ok, unlike_activity, like_activity, object}
 394     else
 395       nil -> {:ok, object}
 396       {:error, error} -> Repo.rollback(error)
 397     end
 398   end
 399
 400   @spec announce(User.t(), Object.t(), String.t() | nil, boolean(), boolean()) ::
 401           {:ok, Activity.t(), Object.t()} | {:error, any()}
 402   def announce(
 403         %User{ap_id: _} = user,
 404         %Object{data: %{"id" => _}} = object,
 405         activity_id \\ nil,
 406         local \\ true,
 407         public \\ true
 408       ) do
 409     with {:ok, result} <-
 410            Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
 411       result
 412     end
 413   end
 414
 415   defp do_announce(user, object, activity_id, local, public) do
 416     with true <- is_announceable?(object, user, public),
 417          object <- Object.get_by_id(object.id),
 418          announce_data <- make_announce_data(user, object, activity_id, public),
 419          {:ok, activity} <- insert(announce_data, local),
 420          {:ok, object} <- add_announce_to_object(activity, object),
 421          :ok <- maybe_federate(activity) do
 422       {:ok, activity, object}
 423     else
 424       false -> {:error, false}
 425       {:error, error} -> Repo.rollback(error)
 426     end
 427   end
 428
 429   @spec unannounce(User.t(), Object.t(), String.t() | nil, boolean()) ::
 430           {:ok, Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
 431   def unannounce(
 432         %User{} = actor,
 433         %Object{} = object,
 434         activity_id \\ nil,
 435         local \\ true
 436       ) do
 437     with {:ok, result} <-
 438            Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
 439       result
 440     end
 441   end
 442
 443   defp do_unannounce(actor, object, activity_id, local) do
 444     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 445          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 446          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 447          :ok <- maybe_federate(unannounce_activity),
 448          {:ok, _activity} <- Repo.delete(announce_activity),
 449          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 450       {:ok, unannounce_activity, object}
 451     else
 452       nil -> {:ok, object}
 453       {:error, error} -> Repo.rollback(error)
 454     end
 455   end
 456
 457   @spec follow(User.t(), User.t(), String.t() | nil, boolean()) ::
 458           {:ok, Activity.t()} | {:error, any()}
 459   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 460     with {:ok, result} <-
 461            Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
 462       result
 463     end
 464   end
 465
 466   defp do_follow(follower, followed, activity_id, local) do
 467     with data <- make_follow_data(follower, followed, activity_id),
 468          {:ok, activity} <- insert(data, local),
 469          :ok <- maybe_federate(activity) do
 470       {:ok, activity}
 471     else
 472       {:error, error} -> Repo.rollback(error)
 473     end
 474   end
 475
 476   @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
 477           {:ok, Activity.t()} | nil | {:error, any()}
 478   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 479     with {:ok, result} <-
 480            Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
 481       result
 482     end
 483   end
 484
 485   defp do_unfollow(follower, followed, activity_id, local) do
 486     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 487          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 488          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 489          {:ok, activity} <- insert(unfollow_data, local),
 490          :ok <- maybe_federate(activity) do
 491       {:ok, activity}
 492     else
 493       nil -> nil
 494       {:error, error} -> Repo.rollback(error)
 495     end
 496   end
 497
 498   @spec delete(User.t() | Object.t(), keyword()) :: {:ok, User.t() | Object.t()} | {:error, any()}
 499   def delete(entity, options \\ []) do
 500     with {:ok, result} <- Repo.transaction(fn -> do_delete(entity, options) end) do
 501       result
 502     end
 503   end
 504
 505   defp do_delete(%User{ap_id: ap_id, follower_address: follower_address} = user, _) do
 506     with data <- %{
 507            "to" => [follower_address],
 508            "type" => "Delete",
 509            "actor" => ap_id,
 510            "object" => %{"type" => "Person", "id" => ap_id}
 511          },
 512          {:ok, activity} <- insert(data, true, true, true),
 513          :ok <- maybe_federate(activity) do
 514       {:ok, user}
 515     end
 516   end
 517
 518   defp do_delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options) do
 519     local = Keyword.get(options, :local, true)
 520     activity_id = Keyword.get(options, :activity_id, nil)
 521     actor = Keyword.get(options, :actor, actor)
 522
 523     user = User.get_cached_by_ap_id(actor)
 524     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 525
 526     with create_activity <- Activity.get_create_by_object_ap_id(id),
 527          data <-
 528            %{
 529              "type" => "Delete",
 530              "actor" => actor,
 531              "object" => id,
 532              "to" => to,
 533              "deleted_activity_id" => create_activity && create_activity.id
 534            }
 535            |> maybe_put("id", activity_id),
 536          {:ok, activity} <- insert(data, local, false),
 537          {:ok, object, _create_activity} <- Object.delete(object),
 538          stream_out_participations(object, user),
 539          _ <- decrease_replies_count_if_reply(object),
 540          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 541          :ok <- maybe_federate(activity) do
 542       {:ok, activity}
 543     else
 544       {:error, error} ->
 545         Repo.rollback(error)
 546     end
 547   end
 548
 549   defp do_delete(%Object{data: %{"type" => "Tombstone", "id" => ap_id}}, _) do
 550     activity =
 551       ap_id
 552       |> Activity.Queries.by_object_id()
 553       |> Activity.Queries.by_type("Delete")
 554       |> Repo.one()
 555
 556     {:ok, activity}
 557   end
 558
 559   @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
 560           {:ok, Activity.t()} | {:error, any()}
 561   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 562     with {:ok, result} <-
 563            Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
 564       result
 565     end
 566   end
 567
 568   defp do_block(blocker, blocked, activity_id, local) do
 569     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
 570     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 571
 572     if unfollow_blocked do
 573       follow_activity = fetch_latest_follow(blocker, blocked)
 574       if follow_activity, do: unfollow(blocker, blocked, nil, local)
 575     end
 576
 577     with true <- outgoing_blocks,
 578          block_data <- make_block_data(blocker, blocked, activity_id),
 579          {:ok, activity} <- insert(block_data, local),
 580          :ok <- maybe_federate(activity) do
 581       {:ok, activity}
 582     else
 583       {:error, error} -> Repo.rollback(error)
 584     end
 585   end
 586
 587   @spec unblock(User.t(), User.t(), String.t() | nil, boolean()) ::
 588           {:ok, Activity.t()} | {:error, any()} | nil
 589   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 590     with {:ok, result} <-
 591            Repo.transaction(fn -> do_unblock(blocker, blocked, activity_id, local) end) do
 592       result
 593     end
 594   end
 595
 596   defp do_unblock(blocker, blocked, activity_id, local) do
 597     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 598          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 599          {:ok, activity} <- insert(unblock_data, local),
 600          :ok <- maybe_federate(activity) do
 601       {:ok, activity}
 602     else
 603       nil -> nil
 604       {:error, error} -> Repo.rollback(error)
 605     end
 606   end
 607
 608   @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
 609   def flag(
 610         %{
 611           actor: actor,
 612           context: _context,
 613           account: account,
 614           statuses: statuses,
 615           content: content
 616         } = params
 617       ) do
 618     # only accept false as false value
 619     local = !(params[:local] == false)
 620     forward = !(params[:forward] == false)
 621
 622     additional = params[:additional] || %{}
 623
 624     additional =
 625       if forward do
 626         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 627       else
 628         Map.merge(additional, %{"to" => [], "cc" => []})
 629       end
 630
 631     with flag_data <- make_flag_data(params, additional),
 632          {:ok, activity} <- insert(flag_data, local),
 633          {:ok, stripped_activity} <- strip_report_status_data(activity),
 634          :ok <- maybe_federate(stripped_activity) do
 635       User.all_superusers()
 636       |> Enum.filter(fn user -> not is_nil(user.email) end)
 637       |> Enum.each(fn superuser ->
 638         superuser
 639         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 640         |> Pleroma.Emails.Mailer.deliver_async()
 641       end)
 642
 643       {:ok, activity}
 644     end
 645   end
 646
 647   @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 648   def move(%User{} = origin, %User{} = target, local \\ true) do
 649     params = %{
 650       "type" => "Move",
 651       "actor" => origin.ap_id,
 652       "object" => origin.ap_id,
 653       "target" => target.ap_id
 654     }
 655
 656     with true <- origin.ap_id in target.also_known_as,
 657          {:ok, activity} <- insert(params, local) do
 658       maybe_federate(activity)
 659
 660       BackgroundWorker.enqueue("move_following", %{
 661         "origin_id" => origin.id,
 662         "target_id" => target.id
 663       })
 664
 665       {:ok, activity}
 666     else
 667       false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
 668       err -> err
 669     end
 670   end
 671
 672   def fetch_activities_for_context_query(context, opts) do
 673     public = [Constants.as_public()]
 674
 675     recipients =
 676       if opts["user"],
 677         do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
 678         else: public
 679
 680     from(activity in Activity)
 681     |> maybe_preload_objects(opts)
 682     |> maybe_preload_bookmarks(opts)
 683     |> maybe_set_thread_muted_field(opts)
 684     |> restrict_blocked(opts)
 685     |> restrict_recipients(recipients, opts["user"])
 686     |> where(
 687       [activity],
 688       fragment(
 689         "?->>'type' = ? and ?->>'context' = ?",
 690         activity.data,
 691         "Create",
 692         activity.data,
 693         ^context
 694       )
 695     )
 696     |> exclude_poll_votes(opts)
 697     |> exclude_id(opts)
 698     |> order_by([activity], desc: activity.id)
 699   end
 700
 701   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 702   def fetch_activities_for_context(context, opts \\ %{}) do
 703     context
 704     |> fetch_activities_for_context_query(opts)
 705     |> Repo.all()
 706   end
 707
 708   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 709           FlakeId.Ecto.CompatType.t() | nil
 710   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 711     context
 712     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 713     |> limit(1)
 714     |> select([a], a.id)
 715     |> Repo.one()
 716   end
 717
 718   @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
 719   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 720     opts = Map.drop(opts, ["user"])
 721
 722     [Constants.as_public()]
 723     |> fetch_activities_query(opts)
 724     |> restrict_unlisted()
 725     |> Pagination.fetch_paginated(opts, pagination)
 726   end
 727
 728   @valid_visibilities ~w[direct unlisted public private]
 729
 730   defp restrict_visibility(query, %{visibility: visibility})
 731        when is_list(visibility) do
 732     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 733       query =
 734         from(
 735           a in query,
 736           where:
 737             fragment(
 738               "activity_visibility(?, ?, ?) = ANY (?)",
 739               a.actor,
 740               a.recipients,
 741               a.data,
 742               ^visibility
 743             )
 744         )
 745
 746       query
 747     else
 748       Logger.error("Could not restrict visibility to #{visibility}")
 749     end
 750   end
 751
 752   defp restrict_visibility(query, %{visibility: visibility})
 753        when visibility in @valid_visibilities do
 754     from(
 755       a in query,
 756       where:
 757         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 758     )
 759   end
 760
 761   defp restrict_visibility(_query, %{visibility: visibility})
 762        when visibility not in @valid_visibilities do
 763     Logger.error("Could not restrict visibility to #{visibility}")
 764   end
 765
 766   defp restrict_visibility(query, _visibility), do: query
 767
 768   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 769        when is_list(visibility) do
 770     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 771       from(
 772         a in query,
 773         where:
 774           not fragment(
 775             "activity_visibility(?, ?, ?) = ANY (?)",
 776             a.actor,
 777             a.recipients,
 778             a.data,
 779             ^visibility
 780           )
 781       )
 782     else
 783       Logger.error("Could not exclude visibility to #{visibility}")
 784       query
 785     end
 786   end
 787
 788   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 789        when visibility in @valid_visibilities do
 790     from(
 791       a in query,
 792       where:
 793         not fragment(
 794           "activity_visibility(?, ?, ?) = ?",
 795           a.actor,
 796           a.recipients,
 797           a.data,
 798           ^visibility
 799         )
 800     )
 801   end
 802
 803   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 804        when visibility not in [nil | @valid_visibilities] do
 805     Logger.error("Could not exclude visibility to #{visibility}")
 806     query
 807   end
 808
 809   defp exclude_visibility(query, _visibility), do: query
 810
 811   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 812     do: query
 813
 814   defp restrict_thread_visibility(
 815          query,
 816          %{"user" => %User{skip_thread_containment: true}},
 817          _
 818        ),
 819        do: query
 820
 821   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 822     from(
 823       a in query,
 824       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 825     )
 826   end
 827
 828   defp restrict_thread_visibility(query, _, _), do: query
 829
 830   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 831     params =
 832       params
 833       |> Map.put("user", reading_user)
 834       |> Map.put("actor_id", user.ap_id)
 835
 836     recipients =
 837       user_activities_recipients(%{
 838         "godmode" => params["godmode"],
 839         "reading_user" => reading_user
 840       })
 841
 842     fetch_activities(recipients, params)
 843     |> Enum.reverse()
 844   end
 845
 846   def fetch_user_activities(user, reading_user, params \\ %{}) do
 847     params =
 848       params
 849       |> Map.put("type", ["Create", "Announce"])
 850       |> Map.put("user", reading_user)
 851       |> Map.put("actor_id", user.ap_id)
 852       |> Map.put("pinned_activity_ids", user.pinned_activities)
 853
 854     params =
 855       if User.blocks?(reading_user, user) do
 856         params
 857       else
 858         params
 859         |> Map.put("blocking_user", reading_user)
 860         |> Map.put("muting_user", reading_user)
 861       end
 862
 863     recipients =
 864       user_activities_recipients(%{
 865         "godmode" => params["godmode"],
 866         "reading_user" => reading_user
 867       })
 868
 869     fetch_activities(recipients, params)
 870     |> Enum.reverse()
 871   end
 872
 873   def fetch_statuses(reading_user, params) do
 874     params =
 875       params
 876       |> Map.put("type", ["Create", "Announce"])
 877
 878     recipients =
 879       user_activities_recipients(%{
 880         "godmode" => params["godmode"],
 881         "reading_user" => reading_user
 882       })
 883
 884     fetch_activities(recipients, params, :offset)
 885     |> Enum.reverse()
 886   end
 887
 888   defp user_activities_recipients(%{"godmode" => true}) do
 889     []
 890   end
 891
 892   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 893     if reading_user do
 894       [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 895     else
 896       [Constants.as_public()]
 897     end
 898   end
 899
 900   defp restrict_since(query, %{"since_id" => ""}), do: query
 901
 902   defp restrict_since(query, %{"since_id" => since_id}) do
 903     from(activity in query, where: activity.id > ^since_id)
 904   end
 905
 906   defp restrict_since(query, _), do: query
 907
 908   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 909     raise "Can't use the child object without preloading!"
 910   end
 911
 912   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 913        when is_list(tag_reject) and tag_reject != [] do
 914     from(
 915       [_activity, object] in query,
 916       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 917     )
 918   end
 919
 920   defp restrict_tag_reject(query, _), do: query
 921
 922   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 923     raise "Can't use the child object without preloading!"
 924   end
 925
 926   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 927        when is_list(tag_all) and tag_all != [] do
 928     from(
 929       [_activity, object] in query,
 930       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 931     )
 932   end
 933
 934   defp restrict_tag_all(query, _), do: query
 935
 936   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 937     raise "Can't use the child object without preloading!"
 938   end
 939
 940   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 941     from(
 942       [_activity, object] in query,
 943       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 944     )
 945   end
 946
 947   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 948     from(
 949       [_activity, object] in query,
 950       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 951     )
 952   end
 953
 954   defp restrict_tag(query, _), do: query
 955
 956   defp restrict_recipients(query, [], _user), do: query
 957
 958   defp restrict_recipients(query, recipients, nil) do
 959     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 960   end
 961
 962   defp restrict_recipients(query, recipients, user) do
 963     from(
 964       activity in query,
 965       where: fragment("? && ?", ^recipients, activity.recipients),
 966       or_where: activity.actor == ^user.ap_id
 967     )
 968   end
 969
 970   defp restrict_local(query, %{"local_only" => true}) do
 971     from(activity in query, where: activity.local == true)
 972   end
 973
 974   defp restrict_local(query, _), do: query
 975
 976   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 977     from(activity in query, where: activity.actor == ^actor_id)
 978   end
 979
 980   defp restrict_actor(query, _), do: query
 981
 982   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 983     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 984   end
 985
 986   defp restrict_type(query, %{"type" => type}) do
 987     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 988   end
 989
 990   defp restrict_type(query, _), do: query
 991
 992   defp restrict_state(query, %{"state" => state}) do
 993     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 994   end
 995
 996   defp restrict_state(query, _), do: query
 997
 998   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 999     from(
1000       [_activity, object] in query,
1001       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
1002     )
1003   end
1004
1005   defp restrict_favorited_by(query, _), do: query
1006
1007   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
1008     raise "Can't use the child object without preloading!"
1009   end
1010
1011   defp restrict_media(query, %{"only_media" => val}) when val in [true, "true", "1"] do
1012     from(
1013       [_activity, object] in query,
1014       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
1015     )
1016   end
1017
1018   defp restrict_media(query, _), do: query
1019
1020   defp restrict_replies(query, %{"exclude_replies" => val}) when val in [true, "true", "1"] do
1021     from(
1022       [_activity, object] in query,
1023       where: fragment("?->>'inReplyTo' is null", object.data)
1024     )
1025   end
1026
1027   defp restrict_replies(query, %{
1028          "reply_filtering_user" => user,
1029          "reply_visibility" => "self"
1030        }) do
1031     from(
1032       [activity, object] in query,
1033       where:
1034         fragment(
1035           "?->>'inReplyTo' is null OR ? = ANY(?)",
1036           object.data,
1037           ^user.ap_id,
1038           activity.recipients
1039         )
1040     )
1041   end
1042
1043   defp restrict_replies(query, %{
1044          "reply_filtering_user" => user,
1045          "reply_visibility" => "following"
1046        }) do
1047     from(
1048       [activity, object] in query,
1049       where:
1050         fragment(
1051           "?->>'inReplyTo' is null OR ? && array_remove(?, ?) OR ? = ?",
1052           object.data,
1053           ^[user.ap_id | User.get_cached_user_friends_ap_ids(user)],
1054           activity.recipients,
1055           activity.actor,
1056           activity.actor,
1057           ^user.ap_id
1058         )
1059     )
1060   end
1061
1062   defp restrict_replies(query, _), do: query
1063
1064   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val in [true, "true", "1"] do
1065     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
1066   end
1067
1068   defp restrict_reblogs(query, _), do: query
1069
1070   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
1071
1072   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
1073     mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
1074
1075     query =
1076       from([activity] in query,
1077         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
1078         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
1079       )
1080
1081     unless opts["skip_preload"] do
1082       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
1083     else
1084       query
1085     end
1086   end
1087
1088   defp restrict_muted(query, _), do: query
1089
1090   defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
1091     blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
1092     domain_blocks = user.domain_blocks || []
1093
1094     following_ap_ids = User.get_friends_ap_ids(user)
1095
1096     query =
1097       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
1098
1099     from(
1100       [activity, object: o] in query,
1101       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
1102       where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
1103       where:
1104         fragment(
1105           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1106           activity.data,
1107           activity.data,
1108           ^blocked_ap_ids
1109         ),
1110       where:
1111         fragment(
1112           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
1113           activity.actor,
1114           ^domain_blocks,
1115           activity.actor,
1116           ^following_ap_ids
1117         ),
1118       where:
1119         fragment(
1120           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
1121           o.data,
1122           ^domain_blocks,
1123           o.data,
1124           ^following_ap_ids
1125         )
1126     )
1127   end
1128
1129   defp restrict_blocked(query, _), do: query
1130
1131   defp restrict_unlisted(query) do
1132     from(
1133       activity in query,
1134       where:
1135         fragment(
1136           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
1137           activity.data,
1138           ^[Constants.as_public()]
1139         )
1140     )
1141   end
1142
1143   # TODO: when all endpoints migrated to OpenAPI compare `pinned` with `true` (boolean) only,
1144   # the same for `restrict_media/2`, `restrict_replies/2`, 'restrict_reblogs/2'
1145   # and `restrict_muted/2`
1146
1147   defp restrict_pinned(query, %{"pinned" => pinned, "pinned_activity_ids" => ids})
1148        when pinned in [true, "true", "1"] do
1149     from(activity in query, where: activity.id in ^ids)
1150   end
1151
1152   defp restrict_pinned(query, _), do: query
1153
1154   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1155     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1156
1157     from(
1158       activity in query,
1159       where:
1160         fragment(
1161           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1162           activity.data,
1163           activity.actor,
1164           ^muted_reblogs
1165         )
1166     )
1167   end
1168
1169   defp restrict_muted_reblogs(query, _), do: query
1170
1171   defp restrict_instance(query, %{"instance" => instance}) do
1172     users =
1173       from(
1174         u in User,
1175         select: u.ap_id,
1176         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1177       )
1178       |> Repo.all()
1179
1180     from(activity in query, where: activity.actor in ^users)
1181   end
1182
1183   defp restrict_instance(query, _), do: query
1184
1185   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1186
1187   defp exclude_poll_votes(query, _) do
1188     if has_named_binding?(query, :object) do
1189       from([activity, object: o] in query,
1190         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1191       )
1192     else
1193       query
1194     end
1195   end
1196
1197   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1198     from(activity in query, where: activity.id != ^id)
1199   end
1200
1201   defp exclude_id(query, _), do: query
1202
1203   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1204
1205   defp maybe_preload_objects(query, _) do
1206     query
1207     |> Activity.with_preloaded_object()
1208   end
1209
1210   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1211
1212   defp maybe_preload_bookmarks(query, opts) do
1213     query
1214     |> Activity.with_preloaded_bookmark(opts["user"])
1215   end
1216
1217   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1218     query
1219     |> Activity.with_preloaded_report_notes()
1220   end
1221
1222   defp maybe_preload_report_notes(query, _), do: query
1223
1224   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1225
1226   defp maybe_set_thread_muted_field(query, opts) do
1227     query
1228     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1229   end
1230
1231   defp maybe_order(query, %{order: :desc}) do
1232     query
1233     |> order_by(desc: :id)
1234   end
1235
1236   defp maybe_order(query, %{order: :asc}) do
1237     query
1238     |> order_by(asc: :id)
1239   end
1240
1241   defp maybe_order(query, _), do: query
1242
1243   defp fetch_activities_query_ap_ids_ops(opts) do
1244     source_user = opts["muting_user"]
1245     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1246
1247     ap_id_relationships =
1248       ap_id_relationships ++
1249         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1250           [:block]
1251         else
1252           []
1253         end
1254
1255     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1256
1257     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1258     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1259
1260     restrict_muted_reblogs_opts =
1261       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1262
1263     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1264   end
1265
1266   def fetch_activities_query(recipients, opts \\ %{}) do
1267     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1268       fetch_activities_query_ap_ids_ops(opts)
1269
1270     config = %{
1271       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1272     }
1273
1274     Activity
1275     |> maybe_preload_objects(opts)
1276     |> maybe_preload_bookmarks(opts)
1277     |> maybe_preload_report_notes(opts)
1278     |> maybe_set_thread_muted_field(opts)
1279     |> maybe_order(opts)
1280     |> restrict_recipients(recipients, opts["user"])
1281     |> restrict_replies(opts)
1282     |> restrict_tag(opts)
1283     |> restrict_tag_reject(opts)
1284     |> restrict_tag_all(opts)
1285     |> restrict_since(opts)
1286     |> restrict_local(opts)
1287     |> restrict_actor(opts)
1288     |> restrict_type(opts)
1289     |> restrict_state(opts)
1290     |> restrict_favorited_by(opts)
1291     |> restrict_blocked(restrict_blocked_opts)
1292     |> restrict_muted(restrict_muted_opts)
1293     |> restrict_media(opts)
1294     |> restrict_visibility(opts)
1295     |> restrict_thread_visibility(opts, config)
1296     |> restrict_reblogs(opts)
1297     |> restrict_pinned(opts)
1298     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1299     |> restrict_instance(opts)
1300     |> Activity.restrict_deactivated_users()
1301     |> exclude_poll_votes(opts)
1302     |> exclude_visibility(opts)
1303   end
1304
1305   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1306     list_memberships = Pleroma.List.memberships(opts["user"])
1307
1308     fetch_activities_query(recipients ++ list_memberships, opts)
1309     |> Pagination.fetch_paginated(opts, pagination)
1310     |> Enum.reverse()
1311     |> maybe_update_cc(list_memberships, opts["user"])
1312   end
1313
1314   @doc """
1315   Fetch favorites activities of user with order by sort adds to favorites
1316   """
1317   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1318   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1319     user.ap_id
1320     |> Activity.Queries.by_actor()
1321     |> Activity.Queries.by_type("Like")
1322     |> Activity.with_joined_object()
1323     |> Object.with_joined_activity()
1324     |> select([_like, object, activity], %{activity | object: object})
1325     |> order_by([like, _, _], desc: like.id)
1326     |> Pagination.fetch_paginated(
1327       Map.merge(params, %{"skip_order" => true}),
1328       pagination,
1329       :object_activity
1330     )
1331   end
1332
1333   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1334        when is_list(list_memberships) and length(list_memberships) > 0 do
1335     Enum.map(activities, fn
1336       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1337         if Enum.any?(bcc, &(&1 in list_memberships)) do
1338           update_in(activity.data["cc"], &[user_ap_id | &1])
1339         else
1340           activity
1341         end
1342
1343       activity ->
1344         activity
1345     end)
1346   end
1347
1348   defp maybe_update_cc(activities, _, _), do: activities
1349
1350   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1351     from(activity in query,
1352       where:
1353         fragment("? && ?", activity.recipients, ^recipients) or
1354           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1355              ^Constants.as_public() in activity.recipients)
1356     )
1357   end
1358
1359   def fetch_activities_bounded(
1360         recipients,
1361         recipients_with_public,
1362         opts \\ %{},
1363         pagination \\ :keyset
1364       ) do
1365     fetch_activities_query([], opts)
1366     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1367     |> Pagination.fetch_paginated(opts, pagination)
1368     |> Enum.reverse()
1369   end
1370
1371   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1372   def upload(file, opts \\ []) do
1373     with {:ok, data} <- Upload.store(file, opts) do
1374       obj_data =
1375         if opts[:actor] do
1376           Map.put(data, "actor", opts[:actor])
1377         else
1378           data
1379         end
1380
1381       Repo.insert(%Object{data: obj_data})
1382     end
1383   end
1384
1385   @spec get_actor_url(any()) :: binary() | nil
1386   defp get_actor_url(url) when is_binary(url), do: url
1387   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1388
1389   defp get_actor_url(url) when is_list(url) do
1390     url
1391     |> List.first()
1392     |> get_actor_url()
1393   end
1394
1395   defp get_actor_url(_url), do: nil
1396
1397   defp object_to_user_data(data) do
1398     avatar =
1399       data["icon"]["url"] &&
1400         %{
1401           "type" => "Image",
1402           "url" => [%{"href" => data["icon"]["url"]}]
1403         }
1404
1405     banner =
1406       data["image"]["url"] &&
1407         %{
1408           "type" => "Image",
1409           "url" => [%{"href" => data["image"]["url"]}]
1410         }
1411
1412     fields =
1413       data
1414       |> Map.get("attachment", [])
1415       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1416       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1417
1418     emojis =
1419       data
1420       |> Map.get("tag", [])
1421       |> Enum.filter(fn
1422         %{"type" => "Emoji"} -> true
1423         _ -> false
1424       end)
1425       |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1426         Map.put(acc, String.trim(name, ":"), url)
1427       end)
1428
1429     locked = data["manuallyApprovesFollowers"] || false
1430     data = Transmogrifier.maybe_fix_user_object(data)
1431     discoverable = data["discoverable"] || false
1432     invisible = data["invisible"] || false
1433     actor_type = data["type"] || "Person"
1434
1435     public_key =
1436       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1437         data["publicKey"]["publicKeyPem"]
1438       else
1439         nil
1440       end
1441
1442     shared_inbox =
1443       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1444         data["endpoints"]["sharedInbox"]
1445       else
1446         nil
1447       end
1448
1449     user_data = %{
1450       ap_id: data["id"],
1451       uri: get_actor_url(data["url"]),
1452       ap_enabled: true,
1453       banner: banner,
1454       fields: fields,
1455       emoji: emojis,
1456       locked: locked,
1457       discoverable: discoverable,
1458       invisible: invisible,
1459       avatar: avatar,
1460       name: data["name"],
1461       follower_address: data["followers"],
1462       following_address: data["following"],
1463       bio: data["summary"],
1464       actor_type: actor_type,
1465       also_known_as: Map.get(data, "alsoKnownAs", []),
1466       public_key: public_key,
1467       inbox: data["inbox"],
1468       shared_inbox: shared_inbox
1469     }
1470
1471     # nickname can be nil because of virtual actors
1472     user_data =
1473       if data["preferredUsername"] do
1474         Map.put(
1475           user_data,
1476           :nickname,
1477           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1478         )
1479       else
1480         Map.put(user_data, :nickname, nil)
1481       end
1482
1483     {:ok, user_data}
1484   end
1485
1486   def fetch_follow_information_for_user(user) do
1487     with {:ok, following_data} <-
1488            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1489          {:ok, hide_follows} <- collection_private(following_data),
1490          {:ok, followers_data} <-
1491            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1492          {:ok, hide_followers} <- collection_private(followers_data) do
1493       {:ok,
1494        %{
1495          hide_follows: hide_follows,
1496          follower_count: normalize_counter(followers_data["totalItems"]),
1497          following_count: normalize_counter(following_data["totalItems"]),
1498          hide_followers: hide_followers
1499        }}
1500     else
1501       {:error, _} = e -> e
1502       e -> {:error, e}
1503     end
1504   end
1505
1506   defp normalize_counter(counter) when is_integer(counter), do: counter
1507   defp normalize_counter(_), do: 0
1508
1509   def maybe_update_follow_information(user_data) do
1510     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1511          {_, true} <- {:user_type_check, user_data[:type] in ["Person", "Service"]},
1512          {_, true} <-
1513            {:collections_available,
1514             !!(user_data[:following_address] && user_data[:follower_address])},
1515          {:ok, info} <-
1516            fetch_follow_information_for_user(user_data) do
1517       info = Map.merge(user_data[:info] || %{}, info)
1518
1519       user_data
1520       |> Map.put(:info, info)
1521     else
1522       {:user_type_check, false} ->
1523         user_data
1524
1525       {:collections_available, false} ->
1526         user_data
1527
1528       {:enabled, false} ->
1529         user_data
1530
1531       e ->
1532         Logger.error(
1533           "Follower/Following counter update for #{user_data.ap_id} failed.\n" <> inspect(e)
1534         )
1535
1536         user_data
1537     end
1538   end
1539
1540   defp collection_private(%{"first" => %{"type" => type}})
1541        when type in ["CollectionPage", "OrderedCollectionPage"],
1542        do: {:ok, false}
1543
1544   defp collection_private(%{"first" => first}) do
1545     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1546            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1547       {:ok, false}
1548     else
1549       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1550       {:error, _} = e -> e
1551       e -> {:error, e}
1552     end
1553   end
1554
1555   defp collection_private(_data), do: {:ok, true}
1556
1557   def user_data_from_user_object(data) do
1558     with {:ok, data} <- MRF.filter(data),
1559          {:ok, data} <- object_to_user_data(data) do
1560       {:ok, data}
1561     else
1562       e -> {:error, e}
1563     end
1564   end
1565
1566   def fetch_and_prepare_user_from_ap_id(ap_id) do
1567     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1568          {:ok, data} <- user_data_from_user_object(data),
1569          data <- maybe_update_follow_information(data) do
1570       {:ok, data}
1571     else
1572       {:error, "Object has been deleted"} = e ->
1573         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1574         {:error, e}
1575
1576       e ->
1577         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1578         {:error, e}
1579     end
1580   end
1581
1582   def make_user_from_ap_id(ap_id) do
1583     user = User.get_cached_by_ap_id(ap_id)
1584
1585     if user && !User.ap_enabled?(user) do
1586       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1587     else
1588       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1589         if user do
1590           user
1591           |> User.remote_user_changeset(data)
1592           |> User.update_and_set_cache()
1593         else
1594           data
1595           |> User.remote_user_changeset()
1596           |> Repo.insert()
1597           |> User.set_cache()
1598         end
1599       else
1600         e -> {:error, e}
1601       end
1602     end
1603   end
1604
1605   def make_user_from_nickname(nickname) do
1606     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1607       make_user_from_ap_id(ap_id)
1608     else
1609       _e -> {:error, "No AP id in WebFinger"}
1610     end
1611   end
1612
1613   # filter out broken threads
1614   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1615     entire_thread_visible_for_user?(activity, user)
1616   end
1617
1618   # do post-processing on a specific activity
1619   def contain_activity(%Activity{} = activity, %User{} = user) do
1620     contain_broken_threads(activity, user)
1621   end
1622
1623   def fetch_direct_messages_query do
1624     Activity
1625     |> restrict_type(%{"type" => "Create"})
1626     |> restrict_visibility(%{visibility: "direct"})
1627     |> order_by([activity], asc: activity.id)
1628   end
1629 end