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