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