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

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