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