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

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