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_or_unlisted_activities(map(), Pagination.type()) :: [Activity.t()]
 542   def fetch_public_or_unlisted_activities(opts \\ %{}, pagination \\ :keyset) do
 543     opts = Map.drop(opts, ["user"])
 544
 545     query = fetch_activities_query([Constants.as_public()], opts)
 546
 547     query =
 548       if opts["restrict_unlisted"] do
 549         restrict_unlisted(query)
 550       else
 551         query
 552       end
 553
 554     Pagination.fetch_paginated(query, opts, pagination)
 555   end
 556
 557   @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
 558   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 559     opts
 560     |> Map.put("restrict_unlisted", true)
 561     |> fetch_public_or_unlisted_activities(pagination)
 562   end
 563
 564   @valid_visibilities ~w[direct unlisted public private]
 565
 566   defp restrict_visibility(query, %{visibility: visibility})
 567        when is_list(visibility) do
 568     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 569       query =
 570         from(
 571           a in query,
 572           where:
 573             fragment(
 574               "activity_visibility(?, ?, ?) = ANY (?)",
 575               a.actor,
 576               a.recipients,
 577               a.data,
 578               ^visibility
 579             )
 580         )
 581
 582       query
 583     else
 584       Logger.error("Could not restrict visibility to #{visibility}")
 585     end
 586   end
 587
 588   defp restrict_visibility(query, %{visibility: visibility})
 589        when visibility in @valid_visibilities do
 590     from(
 591       a in query,
 592       where:
 593         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 594     )
 595   end
 596
 597   defp restrict_visibility(_query, %{visibility: visibility})
 598        when visibility not in @valid_visibilities do
 599     Logger.error("Could not restrict visibility to #{visibility}")
 600   end
 601
 602   defp restrict_visibility(query, _visibility), do: query
 603
 604   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 605        when is_list(visibility) do
 606     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 607       from(
 608         a in query,
 609         where:
 610           not fragment(
 611             "activity_visibility(?, ?, ?) = ANY (?)",
 612             a.actor,
 613             a.recipients,
 614             a.data,
 615             ^visibility
 616           )
 617       )
 618     else
 619       Logger.error("Could not exclude visibility to #{visibility}")
 620       query
 621     end
 622   end
 623
 624   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 625        when visibility in @valid_visibilities do
 626     from(
 627       a in query,
 628       where:
 629         not fragment(
 630           "activity_visibility(?, ?, ?) = ?",
 631           a.actor,
 632           a.recipients,
 633           a.data,
 634           ^visibility
 635         )
 636     )
 637   end
 638
 639   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 640        when visibility not in [nil | @valid_visibilities] do
 641     Logger.error("Could not exclude visibility to #{visibility}")
 642     query
 643   end
 644
 645   defp exclude_visibility(query, _visibility), do: query
 646
 647   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 648     do: query
 649
 650   defp restrict_thread_visibility(
 651          query,
 652          %{"user" => %User{skip_thread_containment: true}},
 653          _
 654        ),
 655        do: query
 656
 657   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 658     from(
 659       a in query,
 660       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 661     )
 662   end
 663
 664   defp restrict_thread_visibility(query, _, _), do: query
 665
 666   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 667     params =
 668       params
 669       |> Map.put("user", reading_user)
 670       |> Map.put("actor_id", user.ap_id)
 671
 672     recipients =
 673       user_activities_recipients(%{
 674         "godmode" => params["godmode"],
 675         "reading_user" => reading_user
 676       })
 677
 678     fetch_activities(recipients, params)
 679     |> Enum.reverse()
 680   end
 681
 682   def fetch_user_activities(user, reading_user, params \\ %{}) do
 683     params =
 684       params
 685       |> Map.put("type", ["Create", "Announce"])
 686       |> Map.put("user", reading_user)
 687       |> Map.put("actor_id", user.ap_id)
 688       |> Map.put("pinned_activity_ids", user.pinned_activities)
 689
 690     params =
 691       if User.blocks?(reading_user, user) do
 692         params
 693       else
 694         params
 695         |> Map.put("blocking_user", reading_user)
 696         |> Map.put("muting_user", reading_user)
 697       end
 698
 699     recipients =
 700       user_activities_recipients(%{
 701         "godmode" => params["godmode"],
 702         "reading_user" => reading_user
 703       })
 704
 705     fetch_activities(recipients, params)
 706     |> Enum.reverse()
 707   end
 708
 709   def fetch_statuses(reading_user, params) do
 710     params =
 711       params
 712       |> Map.put("type", ["Create", "Announce"])
 713
 714     recipients =
 715       user_activities_recipients(%{
 716         "godmode" => params["godmode"],
 717         "reading_user" => reading_user
 718       })
 719
 720     fetch_activities(recipients, params, :offset)
 721     |> Enum.reverse()
 722   end
 723
 724   defp user_activities_recipients(%{"godmode" => true}) do
 725     []
 726   end
 727
 728   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 729     if reading_user do
 730       [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 731     else
 732       [Constants.as_public()]
 733     end
 734   end
 735
 736   defp restrict_since(query, %{"since_id" => ""}), do: query
 737
 738   defp restrict_since(query, %{"since_id" => since_id}) do
 739     from(activity in query, where: activity.id > ^since_id)
 740   end
 741
 742   defp restrict_since(query, _), do: query
 743
 744   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 745     raise "Can't use the child object without preloading!"
 746   end
 747
 748   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 749        when is_list(tag_reject) and tag_reject != [] do
 750     from(
 751       [_activity, object] in query,
 752       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 753     )
 754   end
 755
 756   defp restrict_tag_reject(query, _), do: query
 757
 758   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 759     raise "Can't use the child object without preloading!"
 760   end
 761
 762   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 763        when is_list(tag_all) and tag_all != [] do
 764     from(
 765       [_activity, object] in query,
 766       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 767     )
 768   end
 769
 770   defp restrict_tag_all(query, _), do: query
 771
 772   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 773     raise "Can't use the child object without preloading!"
 774   end
 775
 776   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 777     from(
 778       [_activity, object] in query,
 779       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 780     )
 781   end
 782
 783   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 784     from(
 785       [_activity, object] in query,
 786       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 787     )
 788   end
 789
 790   defp restrict_tag(query, _), do: query
 791
 792   defp restrict_recipients(query, [], _user), do: query
 793
 794   defp restrict_recipients(query, recipients, nil) do
 795     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 796   end
 797
 798   defp restrict_recipients(query, recipients, user) do
 799     from(
 800       activity in query,
 801       where: fragment("? && ?", ^recipients, activity.recipients),
 802       or_where: activity.actor == ^user.ap_id
 803     )
 804   end
 805
 806   defp restrict_local(query, %{"local_only" => true}) do
 807     from(activity in query, where: activity.local == true)
 808   end
 809
 810   defp restrict_local(query, _), do: query
 811
 812   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 813     from(activity in query, where: activity.actor == ^actor_id)
 814   end
 815
 816   defp restrict_actor(query, _), do: query
 817
 818   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 819     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 820   end
 821
 822   defp restrict_type(query, %{"type" => type}) do
 823     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 824   end
 825
 826   defp restrict_type(query, _), do: query
 827
 828   defp restrict_state(query, %{"state" => state}) do
 829     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 830   end
 831
 832   defp restrict_state(query, _), do: query
 833
 834   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 835     from(
 836       [_activity, object] in query,
 837       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 838     )
 839   end
 840
 841   defp restrict_favorited_by(query, _), do: query
 842
 843   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 844     raise "Can't use the child object without preloading!"
 845   end
 846
 847   defp restrict_media(query, %{"only_media" => val}) when val in [true, "true", "1"] do
 848     from(
 849       [_activity, object] in query,
 850       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 851     )
 852   end
 853
 854   defp restrict_media(query, _), do: query
 855
 856   defp restrict_replies(query, %{"exclude_replies" => val}) when val in [true, "true", "1"] do
 857     from(
 858       [_activity, object] in query,
 859       where: fragment("?->>'inReplyTo' is null", object.data)
 860     )
 861   end
 862
 863   defp restrict_replies(query, %{
 864          "reply_filtering_user" => user,
 865          "reply_visibility" => "self"
 866        }) do
 867     from(
 868       [activity, object] in query,
 869       where:
 870         fragment(
 871           "?->>'inReplyTo' is null OR ? = ANY(?)",
 872           object.data,
 873           ^user.ap_id,
 874           activity.recipients
 875         )
 876     )
 877   end
 878
 879   defp restrict_replies(query, %{
 880          "reply_filtering_user" => user,
 881          "reply_visibility" => "following"
 882        }) do
 883     from(
 884       [activity, object] in query,
 885       where:
 886         fragment(
 887           "?->>'inReplyTo' is null OR ? && array_remove(?, ?) OR ? = ?",
 888           object.data,
 889           ^[user.ap_id | User.get_cached_user_friends_ap_ids(user)],
 890           activity.recipients,
 891           activity.actor,
 892           activity.actor,
 893           ^user.ap_id
 894         )
 895     )
 896   end
 897
 898   defp restrict_replies(query, _), do: query
 899
 900   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val in [true, "true", "1"] do
 901     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 902   end
 903
 904   defp restrict_reblogs(query, _), do: query
 905
 906   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 907
 908   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
 909     mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
 910
 911     query =
 912       from([activity] in query,
 913         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 914         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 915       )
 916
 917     unless opts["skip_preload"] do
 918       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
 919     else
 920       query
 921     end
 922   end
 923
 924   defp restrict_muted(query, _), do: query
 925
 926   defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
 927     blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
 928     domain_blocks = user.domain_blocks || []
 929
 930     following_ap_ids = User.get_friends_ap_ids(user)
 931
 932     query =
 933       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 934
 935     from(
 936       [activity, object: o] in query,
 937       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
 938       where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
 939       where:
 940         fragment(
 941           "recipients_contain_blocked_domains(?, ?) = false",
 942           activity.recipients,
 943           ^domain_blocks
 944         ),
 945       where:
 946         fragment(
 947           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 948           activity.data,
 949           activity.data,
 950           ^blocked_ap_ids
 951         ),
 952       where:
 953         fragment(
 954           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
 955           activity.actor,
 956           ^domain_blocks,
 957           activity.actor,
 958           ^following_ap_ids
 959         ),
 960       where:
 961         fragment(
 962           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
 963           o.data,
 964           ^domain_blocks,
 965           o.data,
 966           ^following_ap_ids
 967         )
 968     )
 969   end
 970
 971   defp restrict_blocked(query, _), do: query
 972
 973   defp restrict_unlisted(query) do
 974     from(
 975       activity in query,
 976       where:
 977         fragment(
 978           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 979           activity.data,
 980           ^[Constants.as_public()]
 981         )
 982     )
 983   end
 984
 985   # TODO: when all endpoints migrated to OpenAPI compare `pinned` with `true` (boolean) only,
 986   # the same for `restrict_media/2`, `restrict_replies/2`, 'restrict_reblogs/2'
 987   # and `restrict_muted/2`
 988
 989   defp restrict_pinned(query, %{"pinned" => pinned, "pinned_activity_ids" => ids})
 990        when pinned in [true, "true", "1"] do
 991     from(activity in query, where: activity.id in ^ids)
 992   end
 993
 994   defp restrict_pinned(query, _), do: query
 995
 996   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
 997     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
 998
 999     from(
1000       activity in query,
1001       where:
1002         fragment(
1003           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1004           activity.data,
1005           activity.actor,
1006           ^muted_reblogs
1007         )
1008     )
1009   end
1010
1011   defp restrict_muted_reblogs(query, _), do: query
1012
1013   defp restrict_instance(query, %{"instance" => instance}) do
1014     users =
1015       from(
1016         u in User,
1017         select: u.ap_id,
1018         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1019       )
1020       |> Repo.all()
1021
1022     from(activity in query, where: activity.actor in ^users)
1023   end
1024
1025   defp restrict_instance(query, _), do: query
1026
1027   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1028
1029   defp exclude_poll_votes(query, _) do
1030     if has_named_binding?(query, :object) do
1031       from([activity, object: o] in query,
1032         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1033       )
1034     else
1035       query
1036     end
1037   end
1038
1039   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1040     from(activity in query, where: activity.id != ^id)
1041   end
1042
1043   defp exclude_id(query, _), do: query
1044
1045   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1046
1047   defp maybe_preload_objects(query, _) do
1048     query
1049     |> Activity.with_preloaded_object()
1050   end
1051
1052   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1053
1054   defp maybe_preload_bookmarks(query, opts) do
1055     query
1056     |> Activity.with_preloaded_bookmark(opts["user"])
1057   end
1058
1059   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1060     query
1061     |> Activity.with_preloaded_report_notes()
1062   end
1063
1064   defp maybe_preload_report_notes(query, _), do: query
1065
1066   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1067
1068   defp maybe_set_thread_muted_field(query, opts) do
1069     query
1070     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1071   end
1072
1073   defp maybe_order(query, %{order: :desc}) do
1074     query
1075     |> order_by(desc: :id)
1076   end
1077
1078   defp maybe_order(query, %{order: :asc}) do
1079     query
1080     |> order_by(asc: :id)
1081   end
1082
1083   defp maybe_order(query, _), do: query
1084
1085   defp fetch_activities_query_ap_ids_ops(opts) do
1086     source_user = opts["muting_user"]
1087     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1088
1089     ap_id_relationships =
1090       ap_id_relationships ++
1091         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1092           [:block]
1093         else
1094           []
1095         end
1096
1097     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1098
1099     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1100     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1101
1102     restrict_muted_reblogs_opts =
1103       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1104
1105     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1106   end
1107
1108   def fetch_activities_query(recipients, opts \\ %{}) do
1109     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1110       fetch_activities_query_ap_ids_ops(opts)
1111
1112     config = %{
1113       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1114     }
1115
1116     Activity
1117     |> maybe_preload_objects(opts)
1118     |> maybe_preload_bookmarks(opts)
1119     |> maybe_preload_report_notes(opts)
1120     |> maybe_set_thread_muted_field(opts)
1121     |> maybe_order(opts)
1122     |> restrict_recipients(recipients, opts["user"])
1123     |> restrict_replies(opts)
1124     |> restrict_tag(opts)
1125     |> restrict_tag_reject(opts)
1126     |> restrict_tag_all(opts)
1127     |> restrict_since(opts)
1128     |> restrict_local(opts)
1129     |> restrict_actor(opts)
1130     |> restrict_type(opts)
1131     |> restrict_state(opts)
1132     |> restrict_favorited_by(opts)
1133     |> restrict_blocked(restrict_blocked_opts)
1134     |> restrict_muted(restrict_muted_opts)
1135     |> restrict_media(opts)
1136     |> restrict_visibility(opts)
1137     |> restrict_thread_visibility(opts, config)
1138     |> restrict_reblogs(opts)
1139     |> restrict_pinned(opts)
1140     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1141     |> restrict_instance(opts)
1142     |> Activity.restrict_deactivated_users()
1143     |> exclude_poll_votes(opts)
1144     |> exclude_visibility(opts)
1145   end
1146
1147   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1148     list_memberships = Pleroma.List.memberships(opts["user"])
1149
1150     fetch_activities_query(recipients ++ list_memberships, opts)
1151     |> Pagination.fetch_paginated(opts, pagination)
1152     |> Enum.reverse()
1153     |> maybe_update_cc(list_memberships, opts["user"])
1154   end
1155
1156   @doc """
1157   Fetch favorites activities of user with order by sort adds to favorites
1158   """
1159   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1160   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1161     user.ap_id
1162     |> Activity.Queries.by_actor()
1163     |> Activity.Queries.by_type("Like")
1164     |> Activity.with_joined_object()
1165     |> Object.with_joined_activity()
1166     |> select([_like, object, activity], %{activity | object: object})
1167     |> order_by([like, _, _], desc_nulls_last: like.id)
1168     |> Pagination.fetch_paginated(
1169       Map.merge(params, %{"skip_order" => true}),
1170       pagination,
1171       :object_activity
1172     )
1173   end
1174
1175   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1176        when is_list(list_memberships) and length(list_memberships) > 0 do
1177     Enum.map(activities, fn
1178       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1179         if Enum.any?(bcc, &(&1 in list_memberships)) do
1180           update_in(activity.data["cc"], &[user_ap_id | &1])
1181         else
1182           activity
1183         end
1184
1185       activity ->
1186         activity
1187     end)
1188   end
1189
1190   defp maybe_update_cc(activities, _, _), do: activities
1191
1192   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1193     from(activity in query,
1194       where:
1195         fragment("? && ?", activity.recipients, ^recipients) or
1196           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1197              ^Constants.as_public() in activity.recipients)
1198     )
1199   end
1200
1201   def fetch_activities_bounded(
1202         recipients,
1203         recipients_with_public,
1204         opts \\ %{},
1205         pagination \\ :keyset
1206       ) do
1207     fetch_activities_query([], opts)
1208     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1209     |> Pagination.fetch_paginated(opts, pagination)
1210     |> Enum.reverse()
1211   end
1212
1213   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1214   def upload(file, opts \\ []) do
1215     with {:ok, data} <- Upload.store(file, opts) do
1216       obj_data =
1217         if opts[:actor] do
1218           Map.put(data, "actor", opts[:actor])
1219         else
1220           data
1221         end
1222
1223       Repo.insert(%Object{data: obj_data})
1224     end
1225   end
1226
1227   @spec get_actor_url(any()) :: binary() | nil
1228   defp get_actor_url(url) when is_binary(url), do: url
1229   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1230
1231   defp get_actor_url(url) when is_list(url) do
1232     url
1233     |> List.first()
1234     |> get_actor_url()
1235   end
1236
1237   defp get_actor_url(_url), do: nil
1238
1239   defp object_to_user_data(data) do
1240     avatar =
1241       data["icon"]["url"] &&
1242         %{
1243           "type" => "Image",
1244           "url" => [%{"href" => data["icon"]["url"]}]
1245         }
1246
1247     banner =
1248       data["image"]["url"] &&
1249         %{
1250           "type" => "Image",
1251           "url" => [%{"href" => data["image"]["url"]}]
1252         }
1253
1254     fields =
1255       data
1256       |> Map.get("attachment", [])
1257       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1258       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1259
1260     emojis =
1261       data
1262       |> Map.get("tag", [])
1263       |> Enum.filter(fn
1264         %{"type" => "Emoji"} -> true
1265         _ -> false
1266       end)
1267       |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1268         Map.put(acc, String.trim(name, ":"), url)
1269       end)
1270
1271     locked = data["manuallyApprovesFollowers"] || false
1272     data = Transmogrifier.maybe_fix_user_object(data)
1273     discoverable = data["discoverable"] || false
1274     invisible = data["invisible"] || false
1275     actor_type = data["type"] || "Person"
1276
1277     public_key =
1278       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1279         data["publicKey"]["publicKeyPem"]
1280       else
1281         nil
1282       end
1283
1284     shared_inbox =
1285       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1286         data["endpoints"]["sharedInbox"]
1287       else
1288         nil
1289       end
1290
1291     user_data = %{
1292       ap_id: data["id"],
1293       uri: get_actor_url(data["url"]),
1294       ap_enabled: true,
1295       banner: banner,
1296       fields: fields,
1297       emoji: emojis,
1298       locked: locked,
1299       discoverable: discoverable,
1300       invisible: invisible,
1301       avatar: avatar,
1302       name: data["name"],
1303       follower_address: data["followers"],
1304       following_address: data["following"],
1305       bio: data["summary"],
1306       actor_type: actor_type,
1307       also_known_as: Map.get(data, "alsoKnownAs", []),
1308       public_key: public_key,
1309       inbox: data["inbox"],
1310       shared_inbox: shared_inbox
1311     }
1312
1313     # nickname can be nil because of virtual actors
1314     user_data =
1315       if data["preferredUsername"] do
1316         Map.put(
1317           user_data,
1318           :nickname,
1319           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1320         )
1321       else
1322         Map.put(user_data, :nickname, nil)
1323       end
1324
1325     {:ok, user_data}
1326   end
1327
1328   def fetch_follow_information_for_user(user) do
1329     with {:ok, following_data} <-
1330            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1331          {:ok, hide_follows} <- collection_private(following_data),
1332          {:ok, followers_data} <-
1333            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1334          {:ok, hide_followers} <- collection_private(followers_data) do
1335       {:ok,
1336        %{
1337          hide_follows: hide_follows,
1338          follower_count: normalize_counter(followers_data["totalItems"]),
1339          following_count: normalize_counter(following_data["totalItems"]),
1340          hide_followers: hide_followers
1341        }}
1342     else
1343       {:error, _} = e -> e
1344       e -> {:error, e}
1345     end
1346   end
1347
1348   defp normalize_counter(counter) when is_integer(counter), do: counter
1349   defp normalize_counter(_), do: 0
1350
1351   def maybe_update_follow_information(user_data) do
1352     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1353          {_, true} <- {:user_type_check, user_data[:type] in ["Person", "Service"]},
1354          {_, true} <-
1355            {:collections_available,
1356             !!(user_data[:following_address] && user_data[:follower_address])},
1357          {:ok, info} <-
1358            fetch_follow_information_for_user(user_data) do
1359       info = Map.merge(user_data[:info] || %{}, info)
1360
1361       user_data
1362       |> Map.put(:info, info)
1363     else
1364       {:user_type_check, false} ->
1365         user_data
1366
1367       {:collections_available, false} ->
1368         user_data
1369
1370       {:enabled, false} ->
1371         user_data
1372
1373       e ->
1374         Logger.error(
1375           "Follower/Following counter update for #{user_data.ap_id} failed.\n" <> inspect(e)
1376         )
1377
1378         user_data
1379     end
1380   end
1381
1382   defp collection_private(%{"first" => %{"type" => type}})
1383        when type in ["CollectionPage", "OrderedCollectionPage"],
1384        do: {:ok, false}
1385
1386   defp collection_private(%{"first" => first}) do
1387     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1388            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1389       {:ok, false}
1390     else
1391       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1392       {:error, _} = e -> e
1393       e -> {:error, e}
1394     end
1395   end
1396
1397   defp collection_private(_data), do: {:ok, true}
1398
1399   def user_data_from_user_object(data) do
1400     with {:ok, data} <- MRF.filter(data),
1401          {:ok, data} <- object_to_user_data(data) do
1402       {:ok, data}
1403     else
1404       e -> {:error, e}
1405     end
1406   end
1407
1408   def fetch_and_prepare_user_from_ap_id(ap_id) do
1409     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1410          {:ok, data} <- user_data_from_user_object(data),
1411          data <- maybe_update_follow_information(data) do
1412       {:ok, data}
1413     else
1414       {:error, "Object has been deleted"} = e ->
1415         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1416         {:error, e}
1417
1418       e ->
1419         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1420         {:error, e}
1421     end
1422   end
1423
1424   def make_user_from_ap_id(ap_id) do
1425     user = User.get_cached_by_ap_id(ap_id)
1426
1427     if user && !User.ap_enabled?(user) do
1428       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1429     else
1430       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1431         if user do
1432           user
1433           |> User.remote_user_changeset(data)
1434           |> User.update_and_set_cache()
1435         else
1436           data
1437           |> User.remote_user_changeset()
1438           |> Repo.insert()
1439           |> User.set_cache()
1440         end
1441       else
1442         e -> {:error, e}
1443       end
1444     end
1445   end
1446
1447   def make_user_from_nickname(nickname) do
1448     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1449       make_user_from_ap_id(ap_id)
1450     else
1451       _e -> {:error, "No AP id in WebFinger"}
1452     end
1453   end
1454
1455   # filter out broken threads
1456   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1457     entire_thread_visible_for_user?(activity, user)
1458   end
1459
1460   # do post-processing on a specific activity
1461   def contain_activity(%Activity{} = activity, %User{} = user) do
1462     contain_broken_threads(activity, user)
1463   end
1464
1465   def fetch_direct_messages_query do
1466     Activity
1467     |> restrict_type(%{"type" => "Create"})
1468     |> restrict_visibility(%{visibility: "direct"})
1469     |> order_by([activity], asc: activity.id)
1470   end
1471 end