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