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 Video Event Article]
  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, _} = e ->
 158         {:error, e}
 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:
 845         fragment(
 846           "((not (? && ?)) or ? = ?)",
 847           activity.recipients,
 848           ^blocked_ap_ids,
 849           activity.actor,
 850           ^user.ap_id
 851         ),
 852       where:
 853         fragment(
 854           "recipients_contain_blocked_domains(?, ?) = false",
 855           activity.recipients,
 856           ^domain_blocks
 857         ),
 858       where:
 859         fragment(
 860           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 861           activity.data,
 862           activity.data,
 863           ^blocked_ap_ids
 864         ),
 865       where:
 866         fragment(
 867           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
 868           activity.actor,
 869           ^domain_blocks,
 870           activity.actor,
 871           ^following_ap_ids
 872         ),
 873       where:
 874         fragment(
 875           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
 876           o.data,
 877           ^domain_blocks,
 878           o.data,
 879           ^following_ap_ids
 880         )
 881     )
 882   end
 883
 884   defp restrict_blocked(query, _), do: query
 885
 886   defp restrict_unlisted(query, %{restrict_unlisted: true}) do
 887     from(
 888       activity in query,
 889       where:
 890         fragment(
 891           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 892           activity.data,
 893           ^[Constants.as_public()]
 894         )
 895     )
 896   end
 897
 898   defp restrict_unlisted(query, _), do: query
 899
 900   defp restrict_pinned(query, %{pinned: true, pinned_activity_ids: ids}) do
 901     from(activity in query, where: activity.id in ^ids)
 902   end
 903
 904   defp restrict_pinned(query, _), do: query
 905
 906   defp restrict_muted_reblogs(query, %{muting_user: %User{} = user} = opts) do
 907     muted_reblogs = opts[:reblog_muted_users_ap_ids] || User.reblog_muted_users_ap_ids(user)
 908
 909     from(
 910       activity in query,
 911       where:
 912         fragment(
 913           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 914           activity.data,
 915           activity.actor,
 916           ^muted_reblogs
 917         )
 918     )
 919   end
 920
 921   defp restrict_muted_reblogs(query, _), do: query
 922
 923   defp restrict_instance(query, %{instance: instance}) do
 924     users =
 925       from(
 926         u in User,
 927         select: u.ap_id,
 928         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
 929       )
 930       |> Repo.all()
 931
 932     from(activity in query, where: activity.actor in ^users)
 933   end
 934
 935   defp restrict_instance(query, _), do: query
 936
 937   defp restrict_filtered(query, %{user: %User{} = user}) do
 938     case Filter.compose_regex(user) do
 939       nil ->
 940         query
 941
 942       regex ->
 943         from([activity, object] in query,
 944           where:
 945             fragment("not(?->>'content' ~* ?)", object.data, ^regex) or
 946               activity.actor == ^user.ap_id
 947         )
 948     end
 949   end
 950
 951   defp restrict_filtered(query, %{blocking_user: %User{} = user}) do
 952     restrict_filtered(query, %{user: user})
 953   end
 954
 955   defp restrict_filtered(query, _), do: query
 956
 957   defp exclude_poll_votes(query, %{include_poll_votes: true}), do: query
 958
 959   defp exclude_poll_votes(query, _) do
 960     if has_named_binding?(query, :object) do
 961       from([activity, object: o] in query,
 962         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 963       )
 964     else
 965       query
 966     end
 967   end
 968
 969   defp exclude_chat_messages(query, %{include_chat_messages: true}), do: query
 970
 971   defp exclude_chat_messages(query, _) do
 972     if has_named_binding?(query, :object) do
 973       from([activity, object: o] in query,
 974         where: fragment("not(?->>'type' = ?)", o.data, "ChatMessage")
 975       )
 976     else
 977       query
 978     end
 979   end
 980
 981   defp exclude_invisible_actors(query, %{invisible_actors: true}), do: query
 982
 983   defp exclude_invisible_actors(query, _opts) do
 984     invisible_ap_ids =
 985       User.Query.build(%{invisible: true, select: [:ap_id]})
 986       |> Repo.all()
 987       |> Enum.map(fn %{ap_id: ap_id} -> ap_id end)
 988
 989     from([activity] in query, where: activity.actor not in ^invisible_ap_ids)
 990   end
 991
 992   defp exclude_id(query, %{exclude_id: id}) when is_binary(id) do
 993     from(activity in query, where: activity.id != ^id)
 994   end
 995
 996   defp exclude_id(query, _), do: query
 997
 998   defp maybe_preload_objects(query, %{skip_preload: true}), do: query
 999
1000   defp maybe_preload_objects(query, _) do
1001     query
1002     |> Activity.with_preloaded_object()
1003   end
1004
1005   defp maybe_preload_bookmarks(query, %{skip_preload: true}), do: query
1006
1007   defp maybe_preload_bookmarks(query, opts) do
1008     query
1009     |> Activity.with_preloaded_bookmark(opts[:user])
1010   end
1011
1012   defp maybe_preload_report_notes(query, %{preload_report_notes: true}) do
1013     query
1014     |> Activity.with_preloaded_report_notes()
1015   end
1016
1017   defp maybe_preload_report_notes(query, _), do: query
1018
1019   defp maybe_set_thread_muted_field(query, %{skip_preload: true}), do: query
1020
1021   defp maybe_set_thread_muted_field(query, opts) do
1022     query
1023     |> Activity.with_set_thread_muted_field(opts[:muting_user] || opts[:user])
1024   end
1025
1026   defp maybe_order(query, %{order: :desc}) do
1027     query
1028     |> order_by(desc: :id)
1029   end
1030
1031   defp maybe_order(query, %{order: :asc}) do
1032     query
1033     |> order_by(asc: :id)
1034   end
1035
1036   defp maybe_order(query, _), do: query
1037
1038   defp fetch_activities_query_ap_ids_ops(opts) do
1039     source_user = opts[:muting_user]
1040     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1041
1042     ap_id_relationships =
1043       if opts[:blocking_user] && opts[:blocking_user] == source_user do
1044         [:block | ap_id_relationships]
1045       else
1046         ap_id_relationships
1047       end
1048
1049     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1050
1051     restrict_blocked_opts = Map.merge(%{blocked_users_ap_ids: preloaded_ap_ids[:block]}, opts)
1052     restrict_muted_opts = Map.merge(%{muted_users_ap_ids: preloaded_ap_ids[:mute]}, opts)
1053
1054     restrict_muted_reblogs_opts =
1055       Map.merge(%{reblog_muted_users_ap_ids: preloaded_ap_ids[:reblog_mute]}, opts)
1056
1057     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1058   end
1059
1060   def fetch_activities_query(recipients, opts \\ %{}) do
1061     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1062       fetch_activities_query_ap_ids_ops(opts)
1063
1064     config = %{
1065       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1066     }
1067
1068     Activity
1069     |> maybe_preload_objects(opts)
1070     |> maybe_preload_bookmarks(opts)
1071     |> maybe_preload_report_notes(opts)
1072     |> maybe_set_thread_muted_field(opts)
1073     |> maybe_order(opts)
1074     |> restrict_recipients(recipients, opts[:user])
1075     |> restrict_replies(opts)
1076     |> restrict_tag(opts)
1077     |> restrict_tag_reject(opts)
1078     |> restrict_tag_all(opts)
1079     |> restrict_since(opts)
1080     |> restrict_local(opts)
1081     |> restrict_actor(opts)
1082     |> restrict_type(opts)
1083     |> restrict_state(opts)
1084     |> restrict_favorited_by(opts)
1085     |> restrict_blocked(restrict_blocked_opts)
1086     |> restrict_muted(restrict_muted_opts)
1087     |> restrict_filtered(opts)
1088     |> restrict_media(opts)
1089     |> restrict_visibility(opts)
1090     |> restrict_thread_visibility(opts, config)
1091     |> restrict_reblogs(opts)
1092     |> restrict_pinned(opts)
1093     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1094     |> restrict_instance(opts)
1095     |> restrict_announce_object_actor(opts)
1096     |> restrict_filtered(opts)
1097     |> Activity.restrict_deactivated_users()
1098     |> exclude_poll_votes(opts)
1099     |> exclude_chat_messages(opts)
1100     |> exclude_invisible_actors(opts)
1101     |> exclude_visibility(opts)
1102   end
1103
1104   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1105     list_memberships = Pleroma.List.memberships(opts[:user])
1106
1107     fetch_activities_query(recipients ++ list_memberships, opts)
1108     |> Pagination.fetch_paginated(opts, pagination)
1109     |> Enum.reverse()
1110     |> maybe_update_cc(list_memberships, opts[:user])
1111   end
1112
1113   @doc """
1114   Fetch favorites activities of user with order by sort adds to favorites
1115   """
1116   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1117   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1118     user.ap_id
1119     |> Activity.Queries.by_actor()
1120     |> Activity.Queries.by_type("Like")
1121     |> Activity.with_joined_object()
1122     |> Object.with_joined_activity()
1123     |> select([like, object, activity], %{activity | object: object, pagination_id: like.id})
1124     |> order_by([like, _, _], desc_nulls_last: like.id)
1125     |> Pagination.fetch_paginated(
1126       Map.merge(params, %{skip_order: true}),
1127       pagination
1128     )
1129   end
1130
1131   defp maybe_update_cc(activities, [_ | _] = list_memberships, %User{ap_id: user_ap_id}) do
1132     Enum.map(activities, fn
1133       %{data: %{"bcc" => [_ | _] = bcc}} = activity ->
1134         if Enum.any?(bcc, &(&1 in list_memberships)) do
1135           update_in(activity.data["cc"], &[user_ap_id | &1])
1136         else
1137           activity
1138         end
1139
1140       activity ->
1141         activity
1142     end)
1143   end
1144
1145   defp maybe_update_cc(activities, _, _), do: activities
1146
1147   defp fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1148     from(activity in query,
1149       where:
1150         fragment("? && ?", activity.recipients, ^recipients) or
1151           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1152              ^Constants.as_public() in activity.recipients)
1153     )
1154   end
1155
1156   def fetch_activities_bounded(
1157         recipients,
1158         recipients_with_public,
1159         opts \\ %{},
1160         pagination \\ :keyset
1161       ) do
1162     fetch_activities_query([], opts)
1163     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1164     |> Pagination.fetch_paginated(opts, pagination)
1165     |> Enum.reverse()
1166   end
1167
1168   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1169   def upload(file, opts \\ []) do
1170     with {:ok, data} <- Upload.store(file, opts) do
1171       obj_data = Maps.put_if_present(data, "actor", opts[:actor])
1172
1173       Repo.insert(%Object{data: obj_data})
1174     end
1175   end
1176
1177   @spec get_actor_url(any()) :: binary() | nil
1178   defp get_actor_url(url) when is_binary(url), do: url
1179   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1180
1181   defp get_actor_url(url) when is_list(url) do
1182     url
1183     |> List.first()
1184     |> get_actor_url()
1185   end
1186
1187   defp get_actor_url(_url), do: nil
1188
1189   defp object_to_user_data(data) do
1190     avatar =
1191       data["icon"]["url"] &&
1192         %{
1193           "type" => "Image",
1194           "url" => [%{"href" => data["icon"]["url"]}]
1195         }
1196
1197     banner =
1198       data["image"]["url"] &&
1199         %{
1200           "type" => "Image",
1201           "url" => [%{"href" => data["image"]["url"]}]
1202         }
1203
1204     fields =
1205       data
1206       |> Map.get("attachment", [])
1207       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1208       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1209
1210     emojis =
1211       data
1212       |> Map.get("tag", [])
1213       |> Enum.filter(fn
1214         %{"type" => "Emoji"} -> true
1215         _ -> false
1216       end)
1217       |> Map.new(fn %{"icon" => %{"url" => url}, "name" => name} ->
1218         {String.trim(name, ":"), url}
1219       end)
1220
1221     locked = data["manuallyApprovesFollowers"] || false
1222     capabilities = data["capabilities"] || %{}
1223     accepts_chat_messages = capabilities["acceptsChatMessages"]
1224     data = Transmogrifier.maybe_fix_user_object(data)
1225     discoverable = data["discoverable"] || false
1226     invisible = data["invisible"] || false
1227     actor_type = data["type"] || "Person"
1228
1229     public_key =
1230       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1231         data["publicKey"]["publicKeyPem"]
1232       else
1233         nil
1234       end
1235
1236     shared_inbox =
1237       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1238         data["endpoints"]["sharedInbox"]
1239       else
1240         nil
1241       end
1242
1243     user_data = %{
1244       ap_id: data["id"],
1245       uri: get_actor_url(data["url"]),
1246       ap_enabled: true,
1247       banner: banner,
1248       fields: fields,
1249       emoji: emojis,
1250       locked: locked,
1251       discoverable: discoverable,
1252       invisible: invisible,
1253       avatar: avatar,
1254       name: data["name"],
1255       follower_address: data["followers"],
1256       following_address: data["following"],
1257       bio: data["summary"] || "",
1258       actor_type: actor_type,
1259       also_known_as: Map.get(data, "alsoKnownAs", []),
1260       public_key: public_key,
1261       inbox: data["inbox"],
1262       shared_inbox: shared_inbox,
1263       accepts_chat_messages: accepts_chat_messages
1264     }
1265
1266     # nickname can be nil because of virtual actors
1267     if data["preferredUsername"] do
1268       Map.put(
1269         user_data,
1270         :nickname,
1271         "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1272       )
1273     else
1274       Map.put(user_data, :nickname, nil)
1275     end
1276   end
1277
1278   def fetch_follow_information_for_user(user) do
1279     with {:ok, following_data} <-
1280            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address,
1281              force_http: true
1282            ),
1283          {:ok, hide_follows} <- collection_private(following_data),
1284          {:ok, followers_data} <-
1285            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address, force_http: true),
1286          {:ok, hide_followers} <- collection_private(followers_data) do
1287       {:ok,
1288        %{
1289          hide_follows: hide_follows,
1290          follower_count: normalize_counter(followers_data["totalItems"]),
1291          following_count: normalize_counter(following_data["totalItems"]),
1292          hide_followers: hide_followers
1293        }}
1294     else
1295       {:error, _} = e -> e
1296       e -> {:error, e}
1297     end
1298   end
1299
1300   defp normalize_counter(counter) when is_integer(counter), do: counter
1301   defp normalize_counter(_), do: 0
1302
1303   def maybe_update_follow_information(user_data) do
1304     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1305          {_, true} <- {:user_type_check, user_data[:type] in ["Person", "Service"]},
1306          {_, true} <-
1307            {:collections_available,
1308             !!(user_data[:following_address] && user_data[:follower_address])},
1309          {:ok, info} <-
1310            fetch_follow_information_for_user(user_data) do
1311       info = Map.merge(user_data[:info] || %{}, info)
1312
1313       user_data
1314       |> Map.put(:info, info)
1315     else
1316       {:user_type_check, false} ->
1317         user_data
1318
1319       {:collections_available, false} ->
1320         user_data
1321
1322       {:enabled, false} ->
1323         user_data
1324
1325       e ->
1326         Logger.error(
1327           "Follower/Following counter update for #{user_data.ap_id} failed.\n" <> inspect(e)
1328         )
1329
1330         user_data
1331     end
1332   end
1333
1334   defp collection_private(%{"first" => %{"type" => type}})
1335        when type in ["CollectionPage", "OrderedCollectionPage"],
1336        do: {:ok, false}
1337
1338   defp collection_private(%{"first" => first}) do
1339     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1340            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1341       {:ok, false}
1342     else
1343       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1344       {:error, _} = e -> e
1345       e -> {:error, e}
1346     end
1347   end
1348
1349   defp collection_private(_data), do: {:ok, true}
1350
1351   def user_data_from_user_object(data) do
1352     with {:ok, data} <- MRF.filter(data) do
1353       {:ok, object_to_user_data(data)}
1354     else
1355       e -> {:error, e}
1356     end
1357   end
1358
1359   def fetch_and_prepare_user_from_ap_id(ap_id, opts \\ []) do
1360     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id, opts),
1361          {:ok, data} <- user_data_from_user_object(data) do
1362       {:ok, maybe_update_follow_information(data)}
1363     else
1364       {:error, "Object has been deleted" = e} ->
1365         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1366         {:error, e}
1367
1368       {:error, {:reject, reason} = e} ->
1369         Logger.info("Rejected user #{ap_id}: #{inspect(reason)}")
1370         {:error, e}
1371
1372       {:error, e} ->
1373         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1374         {:error, e}
1375     end
1376   end
1377
1378   def maybe_handle_clashing_nickname(data) do
1379     with nickname when is_binary(nickname) <- data[:nickname],
1380          %User{} = old_user <- User.get_by_nickname(nickname),
1381          {_, false} <- {:ap_id_comparison, data[:ap_id] == old_user.ap_id} do
1382       Logger.info(
1383         "Found an old user for #{nickname}, the old ap id is #{old_user.ap_id}, new one is #{
1384           data[:ap_id]
1385         }, renaming."
1386       )
1387
1388       old_user
1389       |> User.remote_user_changeset(%{nickname: "#{old_user.id}.#{old_user.nickname}"})
1390       |> User.update_and_set_cache()
1391     else
1392       {:ap_id_comparison, true} ->
1393         Logger.info(
1394           "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."
1395         )
1396
1397       _ ->
1398         nil
1399     end
1400   end
1401
1402   def make_user_from_ap_id(ap_id, opts \\ []) do
1403     user = User.get_cached_by_ap_id(ap_id)
1404
1405     if user && !User.ap_enabled?(user) do
1406       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1407     else
1408       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, opts) do
1409         if user do
1410           user
1411           |> User.remote_user_changeset(data)
1412           |> User.update_and_set_cache()
1413         else
1414           maybe_handle_clashing_nickname(data)
1415
1416           data
1417           |> User.remote_user_changeset()
1418           |> Repo.insert()
1419           |> User.set_cache()
1420         end
1421       end
1422     end
1423   end
1424
1425   def make_user_from_nickname(nickname) do
1426     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1427       make_user_from_ap_id(ap_id)
1428     else
1429       _e -> {:error, "No AP id in WebFinger"}
1430     end
1431   end
1432
1433   # filter out broken threads
1434   defp contain_broken_threads(%Activity{} = activity, %User{} = user) do
1435     entire_thread_visible_for_user?(activity, user)
1436   end
1437
1438   # do post-processing on a specific activity
1439   def contain_activity(%Activity{} = activity, %User{} = user) do
1440     contain_broken_threads(activity, user)
1441   end
1442
1443   def fetch_direct_messages_query do
1444     Activity
1445     |> restrict_type(%{type: "Create"})
1446     |> restrict_visibility(%{visibility: "direct"})
1447     |> order_by([activity], asc: activity.id)
1448   end
1449 end