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