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