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