git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   6   alias Akkoma.Collections
   7   alias Pleroma.Activity
   8   alias Pleroma.Activity.Ir.Topics
   9   alias Pleroma.Config
  10   alias Pleroma.Constants
  11   alias Pleroma.Conversation
  12   alias Pleroma.Conversation.Participation
  13   alias Pleroma.Filter
  14   alias Pleroma.Hashtag
  15   alias Pleroma.Maps
  16   alias Pleroma.Notification
  17   alias Pleroma.Object
  18   alias Pleroma.Object.Containment
  19   alias Pleroma.Object.Fetcher
  20   alias Pleroma.Pagination
  21   alias Pleroma.Repo
  22   alias Pleroma.Upload
  23   alias Pleroma.User
  24   alias Pleroma.Web.ActivityPub.MRF
  25   alias Pleroma.Web.ActivityPub.Transmogrifier
  26   alias Pleroma.Web.Streamer
  27   alias Pleroma.Web.WebFinger
  28   alias Pleroma.Workers.BackgroundWorker
  29   alias Pleroma.Workers.PollWorker
  30
  31   import Ecto.Query
  32   import Pleroma.Web.ActivityPub.Utils
  33   import Pleroma.Web.ActivityPub.Visibility
  34
  35   require Logger
  36   require Pleroma.Constants
  37
  38   @behaviour Pleroma.Web.ActivityPub.ActivityPub.Persisting
  39   @behaviour Pleroma.Web.ActivityPub.ActivityPub.Streaming
  40
  41   defp get_recipients(%{"type" => "Create"} = data) do
  42     to = Map.get(data, "to", [])
  43     cc = Map.get(data, "cc", [])
  44     bcc = Map.get(data, "bcc", [])
  45     actor = Map.get(data, "actor", [])
  46     recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
  47     {recipients, to, cc}
  48   end
  49
  50   defp get_recipients(data) do
  51     to = Map.get(data, "to", [])
  52     cc = Map.get(data, "cc", [])
  53     bcc = Map.get(data, "bcc", [])
  54     recipients = Enum.concat([to, cc, bcc])
  55     {recipients, to, cc}
  56   end
  57
  58   defp check_actor_can_insert(%{"type" => "Delete"}), do: true
  59   defp check_actor_can_insert(%{"type" => "Undo"}), do: true
  60
  61   defp check_actor_can_insert(%{"actor" => actor}) when is_binary(actor) do
  62     case User.get_cached_by_ap_id(actor) do
  63       %User{is_active: true} -> true
  64       _ -> false
  65     end
  66   end
  67
  68   defp check_actor_can_insert(_), do: true
  69
  70   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
  71     limit = Config.get([:instance, :remote_limit])
  72     String.length(content) <= limit
  73   end
  74
  75   defp check_remote_limit(_), do: true
  76
  77   def increase_note_count_if_public(actor, object) do
  78     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  79   end
  80
  81   def decrease_note_count_if_public(actor, object) do
  82     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  83   end
  84
  85   def update_last_status_at_if_public(actor, object) do
  86     if is_public?(object), do: User.update_last_status_at(actor), else: {:ok, actor}
  87   end
  88
  89   defp increase_replies_count_if_reply(%{
  90          "object" => %{"inReplyTo" => reply_ap_id} = object,
  91          "type" => "Create"
  92        }) do
  93     if is_public?(object) do
  94       Object.increase_replies_count(reply_ap_id)
  95     end
  96   end
  97
  98   defp increase_replies_count_if_reply(_create_data), do: :noop
  99
 100   @object_types ~w[ChatMessage Question Answer Audio Video Event Article Note Page]
 101   @impl true
 102   def persist(%{"type" => type} = object, meta) when type in @object_types do
 103     with {:ok, object} <- Object.create(object) do
 104       {:ok, object, meta}
 105     end
 106   end
 107
 108   @impl true
 109   def persist(object, meta) do
 110     with local <- Keyword.fetch!(meta, :local),
 111          {recipients, _, _} <- get_recipients(object),
 112          {:ok, activity} <-
 113            Repo.insert(%Activity{
 114              data: object,
 115              local: local,
 116              recipients: recipients,
 117              actor: object["actor"]
 118            }),
 119          # TODO: add tests for expired activities, when Note type will be supported in new pipeline
 120          {:ok, _} <- maybe_create_activity_expiration(activity) do
 121       {:ok, activity, meta}
 122     end
 123   end
 124
 125   @spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 126   def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
 127     with nil <- Activity.normalize(map),
 128          map <- lazy_put_activity_defaults(map, fake),
 129          {_, true} <- {:actor_check, bypass_actor_check || check_actor_can_insert(map)},
 130          {_, true} <- {:remote_limit_pass, check_remote_limit(map)},
 131          {:ok, map} <- MRF.filter(map),
 132          {recipients, _, _} = get_recipients(map),
 133          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
 134          {:containment, :ok} <- {:containment, Containment.contain_child(map)},
 135          {:ok, map, object} <- insert_full_object(map),
 136          {:ok, activity} <- insert_activity_with_expiration(map, local, recipients) do
 137       # Splice in the child object if we have one.
 138       activity = Maps.put_if_present(activity, :object, object)
 139
 140       ConcurrentLimiter.limit(Pleroma.Web.RichMedia.Helpers, fn ->
 141         Task.start(fn -> Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity) end)
 142       end)
 143
 144       # Add local posts to search index
 145       if local, do: Pleroma.Search.add_to_index(activity)
 146
 147       {:ok, activity}
 148     else
 149       %Activity{} = activity ->
 150         {:ok, activity}
 151
 152       {:actor_check, _} ->
 153         {:error, false}
 154
 155       {:containment, _} = error ->
 156         error
 157
 158       {:error, _} = error ->
 159         error
 160
 161       {:fake, true, map, recipients} ->
 162         activity = %Activity{
 163           data: map,
 164           local: local,
 165           actor: map["actor"],
 166           recipients: recipients,
 167           id: "pleroma:fakeid"
 168         }
 169
 170         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 171         {:ok, activity}
 172
 173       {:remote_limit_pass, _} ->
 174         {:error, :remote_limit}
 175
 176       {:reject, _} = e ->
 177         {:error, e}
 178     end
 179   end
 180
 181   defp insert_activity_with_expiration(data, local, recipients) do
 182     struct = %Activity{
 183       data: data,
 184       local: local,
 185       actor: data["actor"],
 186       recipients: recipients
 187     }
 188
 189     with {:ok, activity} <- Repo.insert(struct) do
 190       maybe_create_activity_expiration(activity)
 191     end
 192   end
 193
 194   def notify_and_stream(activity) do
 195     Notification.create_notifications(activity)
 196
 197     conversation = create_or_bump_conversation(activity, activity.actor)
 198     participations = get_participations(conversation)
 199     stream_out(activity)
 200     stream_out_participations(participations)
 201   end
 202
 203   defp maybe_create_activity_expiration(
 204          %{data: %{"expires_at" => %DateTime{} = expires_at}} = activity
 205        ) do
 206     with {:ok, _job} <-
 207            Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
 208              activity_id: activity.id,
 209              expires_at: expires_at
 210            }) do
 211       {:ok, activity}
 212     end
 213   end
 214
 215   defp maybe_create_activity_expiration(activity), do: {:ok, activity}
 216
 217   defp create_or_bump_conversation(activity, actor) do
 218     with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
 219          %User{} = user <- User.get_cached_by_ap_id(actor) do
 220       Participation.mark_as_read(user, conversation)
 221       {:ok, conversation}
 222     end
 223   end
 224
 225   defp get_participations({:ok, conversation}) do
 226     conversation
 227     |> Repo.preload(:participations, force: true)
 228     |> Map.get(:participations)
 229   end
 230
 231   defp get_participations(_), do: []
 232
 233   def stream_out_participations(participations) do
 234     participations =
 235       participations
 236       |> Repo.preload(:user)
 237
 238     Streamer.stream("participation", participations)
 239   end
 240
 241   @impl true
 242   def stream_out_participations(%Object{data: %{"context" => context}}, user) do
 243     with %Conversation{} = conversation <- Conversation.get_for_ap_id(context) do
 244       conversation = Repo.preload(conversation, :participations)
 245
 246       last_activity_id =
 247         fetch_latest_direct_activity_id_for_context(conversation.ap_id, %{
 248           user: user,
 249           blocking_user: user
 250         })
 251
 252       if last_activity_id do
 253         stream_out_participations(conversation.participations)
 254       end
 255     end
 256   end
 257
 258   @impl true
 259   def stream_out_participations(_, _), do: :noop
 260
 261   @impl true
 262   def stream_out(%Activity{data: %{"type" => data_type}} = activity)
 263       when data_type in ["Create", "Announce", "Delete"] do
 264     activity
 265     |> Topics.get_activity_topics()
 266     |> Streamer.stream(activity)
 267   end
 268
 269   @impl true
 270   def stream_out(_activity) do
 271     :noop
 272   end
 273
 274   @spec create(map(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 275   def create(params, fake \\ false) do
 276     with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
 277       result
 278     end
 279   end
 280
 281   defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
 282     additional = params[:additional] || %{}
 283     # only accept false as false value
 284     local = !(params[:local] == false)
 285     published = params[:published]
 286     quick_insert? = Config.get([:env]) == :benchmark
 287
 288     create_data =
 289       make_create_data(
 290         %{to: to, actor: actor, published: published, context: context, object: object},
 291         additional
 292       )
 293
 294     with {:ok, activity} <- insert(create_data, local, fake),
 295          {:fake, false, activity} <- {:fake, fake, activity},
 296          _ <- increase_replies_count_if_reply(create_data),
 297          {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
 298          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 299          {:ok, _actor} <- update_last_status_at_if_public(actor, activity),
 300          _ <- notify_and_stream(activity),
 301          :ok <- maybe_schedule_poll_notifications(activity),
 302          :ok <- maybe_federate(activity) do
 303       {:ok, activity}
 304     else
 305       {:quick_insert, true, activity} ->
 306         {:ok, activity}
 307
 308       {:fake, true, activity} ->
 309         {:ok, activity}
 310
 311       {:error, message} ->
 312         Repo.rollback(message)
 313     end
 314   end
 315
 316   defp maybe_schedule_poll_notifications(activity) do
 317     PollWorker.schedule_poll_end(activity)
 318     :ok
 319   end
 320
 321   @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
 322   def listen(%{to: to, actor: actor, context: context, object: object} = params) do
 323     additional = params[:additional] || %{}
 324     # only accept false as false value
 325     local = !(params[:local] == false)
 326     published = params[:published]
 327
 328     listen_data =
 329       make_listen_data(
 330         %{to: to, actor: actor, published: published, context: context, object: object},
 331         additional
 332       )
 333
 334     with {:ok, activity} <- insert(listen_data, local),
 335          _ <- notify_and_stream(activity),
 336          :ok <- maybe_federate(activity) do
 337       {:ok, activity}
 338     end
 339   end
 340
 341   @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
 342           {:ok, Activity.t()} | nil | {:error, any()}
 343   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 344     with {:ok, result} <-
 345            Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
 346       result
 347     end
 348   end
 349
 350   defp do_unfollow(follower, followed, activity_id, local) do
 351     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 352          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 353          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 354          {:ok, activity} <- insert(unfollow_data, local),
 355          _ <- notify_and_stream(activity),
 356          :ok <- maybe_federate(activity) do
 357       {:ok, activity}
 358     else
 359       nil -> nil
 360       {:error, error} -> Repo.rollback(error)
 361     end
 362   end
 363
 364   @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
 365   def flag(params) do
 366     with {:ok, result} <- Repo.transaction(fn -> do_flag(params) end) do
 367       result
 368     end
 369   end
 370
 371   defp do_flag(
 372          %{
 373            actor: actor,
 374            context: _context,
 375            account: account,
 376            statuses: statuses,
 377            content: content
 378          } = params
 379        ) do
 380     # only accept false as false value
 381     local = !(params[:local] == false)
 382     forward = !(params[:forward] == false)
 383
 384     additional = params[:additional] || %{}
 385
 386     additional =
 387       if forward do
 388         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 389       else
 390         Map.merge(additional, %{"to" => [], "cc" => []})
 391       end
 392
 393     with flag_data <- make_flag_data(params, additional),
 394          {:ok, activity} <- insert(flag_data, local),
 395          {:ok, stripped_activity} <- strip_report_status_data(activity),
 396          _ <- notify_and_stream(activity),
 397          :ok <-
 398            maybe_federate(stripped_activity) do
 399       User.all_superusers()
 400       |> Enum.filter(fn user -> user.ap_id != actor end)
 401       |> Enum.filter(fn user -> not is_nil(user.email) end)
 402       |> Enum.each(fn superuser ->
 403         superuser
 404         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 405         |> Pleroma.Emails.Mailer.deliver_async()
 406       end)
 407
 408       {:ok, activity}
 409     else
 410       {:error, error} -> Repo.rollback(error)
 411     end
 412   end
 413
 414   @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 415   def move(%User{} = origin, %User{} = target, local \\ true) do
 416     params = %{
 417       "type" => "Move",
 418       "actor" => origin.ap_id,
 419       "object" => origin.ap_id,
 420       "target" => target.ap_id
 421     }
 422
 423     with true <- origin.ap_id in target.also_known_as,
 424          {:ok, activity} <- insert(params, local),
 425          _ <- notify_and_stream(activity) do
 426       maybe_federate(activity)
 427
 428       BackgroundWorker.enqueue("move_following", %{
 429         "origin_id" => origin.id,
 430         "target_id" => target.id
 431       })
 432
 433       {:ok, activity}
 434     else
 435       false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
 436       err -> err
 437     end
 438   end
 439
 440   def fetch_activities_for_context_query(context, opts) do
 441     public = [Constants.as_public()]
 442
 443     recipients =
 444       if opts[:user],
 445         do: [opts[:user].ap_id | User.following(opts[:user])] ++ public,
 446         else: public
 447
 448     from(activity in Activity)
 449     |> maybe_preload_objects(opts)
 450     |> maybe_preload_bookmarks(opts)
 451     |> maybe_set_thread_muted_field(opts)
 452     |> restrict_blocked(opts)
 453     |> restrict_blockers_visibility(opts)
 454     |> restrict_recipients(recipients, opts[:user])
 455     |> restrict_filtered(opts)
 456     |> where(
 457       [activity],
 458       fragment(
 459         "?->>'type' = ? and ?->>'context' = ?",
 460         activity.data,
 461         "Create",
 462         activity.data,
 463         ^context
 464       )
 465     )
 466     |> exclude_poll_votes(opts)
 467     |> exclude_id(opts)
 468     |> order_by([activity], desc: activity.id)
 469   end
 470
 471   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 472   def fetch_activities_for_context(context, opts \\ %{}) do
 473     context
 474     |> fetch_activities_for_context_query(opts)
 475     |> Repo.all()
 476   end
 477
 478   @spec fetch_latest_direct_activity_id_for_context(String.t(), keyword() | map()) ::
 479           FlakeId.Ecto.CompatType.t() | nil
 480   def fetch_latest_direct_activity_id_for_context(context, opts \\ %{}) do
 481     context
 482     |> fetch_activities_for_context_query(Map.merge(%{skip_preload: true}, opts))
 483     |> restrict_visibility(%{visibility: "direct"})
 484     |> limit(1)
 485     |> select([a], a.id)
 486     |> Repo.one()
 487   end
 488
 489   defp fetch_paginated_optimized(query, opts, pagination) do
 490     # Note: tag-filtering funcs may apply "ORDER BY objects.id DESC",
 491     #   and extra sorting on "activities.id DESC NULLS LAST" would worse the query plan
 492     opts = Map.put(opts, :skip_extra_order, true)
 493
 494     Pagination.fetch_paginated(query, opts, pagination)
 495   end
 496
 497   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
 498     list_memberships = Pleroma.List.memberships(opts[:user])
 499
 500     fetch_activities_query(recipients ++ list_memberships, opts)
 501     |> fetch_paginated_optimized(opts, pagination)
 502     |> Enum.reverse()
 503     |> maybe_update_cc(list_memberships, opts[:user])
 504   end
 505
 506   @spec fetch_public_or_unlisted_activities(map(), Pagination.type()) :: [Activity.t()]
 507   def fetch_public_or_unlisted_activities(opts \\ %{}, pagination \\ :keyset) do
 508     opts = Map.delete(opts, :user)
 509
 510     [Constants.as_public()]
 511     |> fetch_activities_query(opts)
 512     |> restrict_unlisted(opts)
 513     |> fetch_paginated_optimized(opts, pagination)
 514   end
 515
 516   @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
 517   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 518     opts
 519     |> Map.put(:restrict_unlisted, true)
 520     |> fetch_public_or_unlisted_activities(pagination)
 521   end
 522
 523   @valid_visibilities ~w[direct unlisted public private]
 524
 525   defp restrict_visibility(query, %{visibility: visibility})
 526        when is_list(visibility) do
 527     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 528       from(
 529         a in query,
 530         where:
 531           fragment(
 532             "activity_visibility(?, ?, ?) = ANY (?)",
 533             a.actor,
 534             a.recipients,
 535             a.data,
 536             ^visibility
 537           )
 538       )
 539     else
 540       Logger.error("Could not restrict visibility to #{visibility}")
 541     end
 542   end
 543
 544   defp restrict_visibility(query, %{visibility: visibility})
 545        when visibility in @valid_visibilities do
 546     from(
 547       a in query,
 548       where:
 549         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 550     )
 551   end
 552
 553   defp restrict_visibility(_query, %{visibility: visibility})
 554        when visibility not in @valid_visibilities do
 555     Logger.error("Could not restrict visibility to #{visibility}")
 556   end
 557
 558   defp restrict_visibility(query, _visibility), do: query
 559
 560   defp exclude_visibility(query, %{exclude_visibilities: visibility})
 561        when is_list(visibility) do
 562     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 563       from(
 564         a in query,
 565         where:
 566           not fragment(
 567             "activity_visibility(?, ?, ?) = ANY (?)",
 568             a.actor,
 569             a.recipients,
 570             a.data,
 571             ^visibility
 572           )
 573       )
 574     else
 575       Logger.error("Could not exclude visibility to #{visibility}")
 576       query
 577     end
 578   end
 579
 580   defp exclude_visibility(query, %{exclude_visibilities: visibility})
 581        when visibility in @valid_visibilities do
 582     from(
 583       a in query,
 584       where:
 585         not fragment(
 586           "activity_visibility(?, ?, ?) = ?",
 587           a.actor,
 588           a.recipients,
 589           a.data,
 590           ^visibility
 591         )
 592     )
 593   end
 594
 595   defp exclude_visibility(query, %{exclude_visibilities: visibility})
 596        when visibility not in [nil | @valid_visibilities] do
 597     Logger.error("Could not exclude visibility to #{visibility}")
 598     query
 599   end
 600
 601   defp exclude_visibility(query, _visibility), do: query
 602
 603   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 604     do: query
 605
 606   defp restrict_thread_visibility(query, %{user: %User{skip_thread_containment: true}}, _),
 607     do: query
 608
 609   defp restrict_thread_visibility(query, %{user: %User{ap_id: ap_id}}, _) do
 610     from(
 611       a in query,
 612       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 613     )
 614   end
 615
 616   defp restrict_thread_visibility(query, _, _), do: query
 617
 618   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 619     params =
 620       params
 621       |> Map.put(:user, reading_user)
 622       |> Map.put(:actor_id, user.ap_id)
 623
 624     %{
 625       godmode: params[:godmode],
 626       reading_user: reading_user
 627     }
 628     |> user_activities_recipients()
 629     |> fetch_activities(params)
 630     |> Enum.reverse()
 631   end
 632
 633   def fetch_user_activities(user, reading_user, params \\ %{})
 634
 635   def fetch_user_activities(user, reading_user, %{total: true} = params) do
 636     result = fetch_activities_for_user(user, reading_user, params)
 637
 638     Keyword.put(result, :items, Enum.reverse(result[:items]))
 639   end
 640
 641   def fetch_user_activities(user, reading_user, params) do
 642     user
 643     |> fetch_activities_for_user(reading_user, params)
 644     |> Enum.reverse()
 645   end
 646
 647   defp fetch_activities_for_user(user, reading_user, params) do
 648     params =
 649       params
 650       |> Map.put(:type, ["Create", "Announce"])
 651       |> Map.put(:user, reading_user)
 652       |> Map.put(:actor_id, user.ap_id)
 653       |> Map.put(:pinned_object_ids, Map.keys(user.pinned_objects))
 654
 655     params =
 656       if User.blocks?(reading_user, user) do
 657         params
 658       else
 659         params
 660         |> Map.put(:blocking_user, reading_user)
 661         |> Map.put(:muting_user, reading_user)
 662       end
 663
 664     pagination_type = Map.get(params, :pagination_type) || :keyset
 665
 666     %{
 667       godmode: params[:godmode],
 668       reading_user: reading_user
 669     }
 670     |> user_activities_recipients()
 671     |> fetch_activities(params, pagination_type)
 672   end
 673
 674   def fetch_statuses(reading_user, %{total: true} = params) do
 675     result = fetch_activities_for_reading_user(reading_user, params)
 676     Keyword.put(result, :items, Enum.reverse(result[:items]))
 677   end
 678
 679   def fetch_statuses(reading_user, params) do
 680     reading_user
 681     |> fetch_activities_for_reading_user(params)
 682     |> Enum.reverse()
 683   end
 684
 685   defp fetch_activities_for_reading_user(reading_user, params) do
 686     params = Map.put(params, :type, ["Create", "Announce"])
 687
 688     %{
 689       godmode: params[:godmode],
 690       reading_user: reading_user
 691     }
 692     |> user_activities_recipients()
 693     |> fetch_activities(params, :offset)
 694   end
 695
 696   defp user_activities_recipients(%{godmode: true}), do: []
 697
 698   defp user_activities_recipients(%{reading_user: reading_user}) do
 699     if reading_user do
 700       [Constants.as_public(), reading_user.ap_id | User.following(reading_user)]
 701     else
 702       [Constants.as_public()]
 703     end
 704   end
 705
 706   defp restrict_announce_object_actor(_query, %{announce_filtering_user: _, skip_preload: true}) do
 707     raise "Can't use the child object without preloading!"
 708   end
 709
 710   defp restrict_announce_object_actor(query, %{announce_filtering_user: %{ap_id: actor}}) do
 711     from(
 712       [activity, object] in query,
 713       where:
 714         fragment(
 715           "?->>'type' != ? or ?->>'actor' != ?",
 716           activity.data,
 717           "Announce",
 718           object.data,
 719           ^actor
 720         )
 721     )
 722   end
 723
 724   defp restrict_announce_object_actor(query, _), do: query
 725
 726   defp restrict_since(query, %{since_id: ""}), do: query
 727
 728   defp restrict_since(query, %{since_id: since_id}) do
 729     from(activity in query, where: activity.id > ^since_id)
 730   end
 731
 732   defp restrict_since(query, _), do: query
 733
 734   defp restrict_embedded_tag_all(_query, %{tag_all: _tag_all, skip_preload: true}) do
 735     raise_on_missing_preload()
 736   end
 737
 738   defp restrict_embedded_tag_all(query, %{tag_all: [_ | _] = tag_all}) do
 739     from(
 740       [_activity, object] in query,
 741       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 742     )
 743   end
 744
 745   defp restrict_embedded_tag_all(query, %{tag_all: tag}) when is_binary(tag) do
 746     restrict_embedded_tag_any(query, %{tag: tag})
 747   end
 748
 749   defp restrict_embedded_tag_all(query, _), do: query
 750
 751   defp restrict_embedded_tag_any(_query, %{tag: _tag, skip_preload: true}) do
 752     raise_on_missing_preload()
 753   end
 754
 755   defp restrict_embedded_tag_any(query, %{tag: [_ | _] = tag_any}) do
 756     from(
 757       [_activity, object] in query,
 758       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag_any)
 759     )
 760   end
 761
 762   defp restrict_embedded_tag_any(query, %{tag: tag}) when is_binary(tag) do
 763     restrict_embedded_tag_any(query, %{tag: [tag]})
 764   end
 765
 766   defp restrict_embedded_tag_any(query, _), do: query
 767
 768   defp restrict_embedded_tag_reject_any(_query, %{tag_reject: _tag_reject, skip_preload: true}) do
 769     raise_on_missing_preload()
 770   end
 771
 772   defp restrict_embedded_tag_reject_any(query, %{tag_reject: [_ | _] = tag_reject}) do
 773     from(
 774       [_activity, object] in query,
 775       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 776     )
 777   end
 778
 779   defp restrict_embedded_tag_reject_any(query, %{tag_reject: tag_reject})
 780        when is_binary(tag_reject) do
 781     restrict_embedded_tag_reject_any(query, %{tag_reject: [tag_reject]})
 782   end
 783
 784   defp restrict_embedded_tag_reject_any(query, _), do: query
 785
 786   defp object_ids_query_for_tags(tags) do
 787     from(hto in "hashtags_objects")
 788     |> join(:inner, [hto], ht in Pleroma.Hashtag, on: hto.hashtag_id == ht.id)
 789     |> where([hto, ht], ht.name in ^tags)
 790     |> select([hto], hto.object_id)
 791     |> distinct([hto], true)
 792   end
 793
 794   defp restrict_hashtag_all(_query, %{tag_all: _tag, skip_preload: true}) do
 795     raise_on_missing_preload()
 796   end
 797
 798   defp restrict_hashtag_all(query, %{tag_all: [single_tag]}) do
 799     restrict_hashtag_any(query, %{tag: single_tag})
 800   end
 801
 802   defp restrict_hashtag_all(query, %{tag_all: [_ | _] = tags}) do
 803     from(
 804       [_activity, object] in query,
 805       where:
 806         fragment(
 807           """
 808           (SELECT array_agg(hashtags.name) FROM hashtags JOIN hashtags_objects
 809             ON hashtags_objects.hashtag_id = hashtags.id WHERE hashtags.name = ANY(?)
 810               AND hashtags_objects.object_id = ?) @> ?
 811           """,
 812           ^tags,
 813           object.id,
 814           ^tags
 815         )
 816     )
 817   end
 818
 819   defp restrict_hashtag_all(query, %{tag_all: tag}) when is_binary(tag) do
 820     restrict_hashtag_all(query, %{tag_all: [tag]})
 821   end
 822
 823   defp restrict_hashtag_all(query, _), do: query
 824
 825   defp restrict_hashtag_any(_query, %{tag: _tag, skip_preload: true}) do
 826     raise_on_missing_preload()
 827   end
 828
 829   defp restrict_hashtag_any(query, %{tag: [_ | _] = tags}) do
 830     hashtag_ids =
 831       from(ht in Hashtag, where: ht.name in ^tags, select: ht.id)
 832       |> Repo.all()
 833
 834     # Note: NO extra ordering should be done on "activities.id desc nulls last" for optimal plan
 835     from(
 836       [_activity, object] in query,
 837       join: hto in "hashtags_objects",
 838       on: hto.object_id == object.id,
 839       where: hto.hashtag_id in ^hashtag_ids,
 840       distinct: [desc: object.id],
 841       order_by: [desc: object.id]
 842     )
 843   end
 844
 845   defp restrict_hashtag_any(query, %{tag: tag}) when is_binary(tag) do
 846     restrict_hashtag_any(query, %{tag: [tag]})
 847   end
 848
 849   defp restrict_hashtag_any(query, _), do: query
 850
 851   defp restrict_hashtag_reject_any(_query, %{tag_reject: _tag_reject, skip_preload: true}) do
 852     raise_on_missing_preload()
 853   end
 854
 855   defp restrict_hashtag_reject_any(query, %{tag_reject: [_ | _] = tags_reject}) do
 856     from(
 857       [_activity, object] in query,
 858       where: object.id not in subquery(object_ids_query_for_tags(tags_reject))
 859     )
 860   end
 861
 862   defp restrict_hashtag_reject_any(query, %{tag_reject: tag_reject}) when is_binary(tag_reject) do
 863     restrict_hashtag_reject_any(query, %{tag_reject: [tag_reject]})
 864   end
 865
 866   defp restrict_hashtag_reject_any(query, _), do: query
 867
 868   defp raise_on_missing_preload do
 869     raise "Can't use the child object without preloading!"
 870   end
 871
 872   defp restrict_recipients(query, [], _user), do: query
 873
 874   defp restrict_recipients(query, recipients, nil) do
 875     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 876   end
 877
 878   defp restrict_recipients(query, recipients, user) do
 879     from(
 880       activity in query,
 881       where: fragment("? && ?", ^recipients, activity.recipients),
 882       or_where: activity.actor == ^user.ap_id
 883     )
 884   end
 885
 886   defp restrict_local(query, %{local_only: true}) do
 887     from(activity in query, where: activity.local == true)
 888   end
 889
 890   defp restrict_local(query, _), do: query
 891
 892   defp restrict_remote(query, %{remote: true}) do
 893     from(activity in query, where: activity.local == false)
 894   end
 895
 896   defp restrict_remote(query, _), do: query
 897
 898   defp restrict_actor(query, %{actor_id: actor_id}) do
 899     from(activity in query, where: activity.actor == ^actor_id)
 900   end
 901
 902   defp restrict_actor(query, _), do: query
 903
 904   defp restrict_type(query, %{type: type}) when is_binary(type) do
 905     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 906   end
 907
 908   defp restrict_type(query, %{type: type}) do
 909     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 910   end
 911
 912   defp restrict_type(query, _), do: query
 913
 914   defp restrict_state(query, %{state: state}) do
 915     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 916   end
 917
 918   defp restrict_state(query, _), do: query
 919
 920   defp restrict_favorited_by(query, %{favorited_by: ap_id}) do
 921     from(
 922       [_activity, object] in query,
 923       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 924     )
 925   end
 926
 927   defp restrict_favorited_by(query, _), do: query
 928
 929   defp restrict_media(_query, %{only_media: _val, skip_preload: true}) do
 930     raise "Can't use the child object without preloading!"
 931   end
 932
 933   defp restrict_media(query, %{only_media: true}) do
 934     from(
 935       [activity, object] in query,
 936       where: fragment("(?)->>'type' = ?", activity.data, "Create"),
 937       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 938     )
 939   end
 940
 941   defp restrict_media(query, _), do: query
 942
 943   defp restrict_replies(query, %{exclude_replies: true}) do
 944     from(
 945       [_activity, object] in query,
 946       where: fragment("?->>'inReplyTo' is null", object.data)
 947     )
 948   end
 949
 950   defp restrict_replies(query, %{
 951          reply_filtering_user: %User{} = user,
 952          reply_visibility: "self"
 953        }) do
 954     from(
 955       [activity, object] in query,
 956       where:
 957         fragment(
 958           "?->>'inReplyTo' is null OR ? = ANY(?)",
 959           object.data,
 960           ^user.ap_id,
 961           activity.recipients
 962         )
 963     )
 964   end
 965
 966   defp restrict_replies(query, %{
 967          reply_filtering_user: %User{} = user,
 968          reply_visibility: "following"
 969        }) do
 970     from(
 971       [activity, object] in query,
 972       where:
 973         fragment(
 974           """
 975           ?->>'type' != 'Create'     -- This isn't a Create
 976           OR ?->>'inReplyTo' is null -- this isn't a reply
 977           OR ? && array_remove(?, ?) -- The recipient is us or one of our friends,
 978                                      -- unless they are the author (because authors
 979                                      -- are also part of the recipients). This leads
 980                                      -- to a bug that self-replies by friends won't
 981                                      -- show up.
 982           OR ? = ?                   -- The actor is us
 983           """,
 984           activity.data,
 985           object.data,
 986           ^[user.ap_id | User.get_cached_user_friends_ap_ids(user)],
 987           activity.recipients,
 988           activity.actor,
 989           activity.actor,
 990           ^user.ap_id
 991         )
 992     )
 993   end
 994
 995   defp restrict_replies(query, _), do: query
 996
 997   defp restrict_reblogs(query, %{exclude_reblogs: true}) do
 998     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 999   end
1000
1001   defp restrict_reblogs(query, _), do: query
1002
1003   defp restrict_muted(query, %{with_muted: true}), do: query
1004
1005   defp restrict_muted(query, %{muting_user: %User{} = user} = opts) do
1006     mutes = opts[:muted_users_ap_ids] || User.muted_users_ap_ids(user)
1007
1008     query =
1009       from([activity] in query,
1010         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
1011         where:
1012           fragment(
1013             "not (?->'to' \\?| ?) or ? = ?",
1014             activity.data,
1015             ^mutes,
1016             activity.actor,
1017             ^user.ap_id
1018           )
1019       )
1020
1021     unless opts[:skip_preload] do
1022       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
1023     else
1024       query
1025     end
1026   end
1027
1028   defp restrict_muted(query, _), do: query
1029
1030   defp restrict_blocked(query, %{blocking_user: %User{} = user} = opts) do
1031     blocked_ap_ids = opts[:blocked_users_ap_ids] || User.blocked_users_ap_ids(user)
1032     domain_blocks = user.domain_blocks || []
1033
1034     following_ap_ids = User.get_friends_ap_ids(user)
1035
1036     query =
1037       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
1038
1039     from(
1040       [activity, object: o] in query,
1041       # You don't block the author
1042       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
1043
1044       # You don't block any recipients, and didn't author the post
1045       where:
1046         fragment(
1047           "((not (? && ?)) or ? = ?)",
1048           activity.recipients,
1049           ^blocked_ap_ids,
1050           activity.actor,
1051           ^user.ap_id
1052         ),
1053
1054       # You don't block the domain of any recipients, and didn't author the post
1055       where:
1056         fragment(
1057           "(recipients_contain_blocked_domains(?, ?) = false) or ? = ?",
1058           activity.recipients,
1059           ^domain_blocks,
1060           activity.actor,
1061           ^user.ap_id
1062         ),
1063
1064       # It's not a boost of a user you block
1065       where:
1066         fragment(
1067           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1068           activity.data,
1069           activity.data,
1070           ^blocked_ap_ids
1071         ),
1072
1073       # You don't block the author's domain, and also don't follow the author
1074       where:
1075         fragment(
1076           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
1077           activity.actor,
1078           ^domain_blocks,
1079           activity.actor,
1080           ^following_ap_ids
1081         ),
1082
1083       # Same as above, but checks the Object
1084       where:
1085         fragment(
1086           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
1087           o.data,
1088           ^domain_blocks,
1089           o.data,
1090           ^following_ap_ids
1091         )
1092     )
1093   end
1094
1095   defp restrict_blocked(query, _), do: query
1096
1097   defp restrict_blockers_visibility(query, %{blocking_user: %User{} = user}) do
1098     if Config.get([:activitypub, :blockers_visible]) == true do
1099       query
1100     else
1101       blocker_ap_ids = User.incoming_relationships_ungrouped_ap_ids(user, [:block])
1102
1103       from(
1104         activity in query,
1105         # The author doesn't block you
1106         where: fragment("not (? = ANY(?))", activity.actor, ^blocker_ap_ids),
1107
1108         # It's not a boost of a user that blocks you
1109         where:
1110           fragment(
1111             "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1112             activity.data,
1113             activity.data,
1114             ^blocker_ap_ids
1115           )
1116       )
1117     end
1118   end
1119
1120   defp restrict_blockers_visibility(query, _), do: query
1121
1122   defp restrict_unlisted(query, %{restrict_unlisted: true}) do
1123     from(
1124       activity in query,
1125       where:
1126         fragment(
1127           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
1128           activity.data,
1129           ^[Constants.as_public()]
1130         )
1131     )
1132   end
1133
1134   defp restrict_unlisted(query, _), do: query
1135
1136   defp restrict_pinned(query, %{pinned: true, pinned_object_ids: ids}) do
1137     from(
1138       [activity, object: o] in query,
1139       where:
1140         fragment(
1141           "(?)->>'type' = 'Create' and coalesce((?)->'object'->>'id', (?)->>'object') = any (?)",
1142           activity.data,
1143           activity.data,
1144           activity.data,
1145           ^ids
1146         )
1147     )
1148   end
1149
1150   defp restrict_pinned(query, _), do: query
1151
1152   defp restrict_muted_reblogs(query, %{muting_user: %User{} = user} = opts) do
1153     muted_reblogs = opts[:reblog_muted_users_ap_ids] || User.reblog_muted_users_ap_ids(user)
1154
1155     from(
1156       activity in query,
1157       where:
1158         fragment(
1159           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1160           activity.data,
1161           activity.actor,
1162           ^muted_reblogs
1163         )
1164     )
1165   end
1166
1167   defp restrict_muted_reblogs(query, _), do: query
1168
1169   defp restrict_instance(query, %{instance: instance}) when is_binary(instance) do
1170     from(
1171       activity in query,
1172       where: fragment("split_part(actor::text, '/'::text, 3) = ?", ^instance)
1173     )
1174   end
1175
1176   defp restrict_instance(query, _), do: query
1177
1178   defp restrict_filtered(query, %{user: %User{} = user}) do
1179     case Filter.compose_regex(user) do
1180       nil ->
1181         query
1182
1183       regex ->
1184         from([activity, object] in query,
1185           where:
1186             fragment("not(?->>'content' ~* ?)", object.data, ^regex) or
1187               activity.actor == ^user.ap_id
1188         )
1189     end
1190   end
1191
1192   defp restrict_filtered(query, %{blocking_user: %User{} = user}) do
1193     restrict_filtered(query, %{user: user})
1194   end
1195
1196   defp restrict_filtered(query, _), do: query
1197
1198   defp exclude_poll_votes(query, %{include_poll_votes: true}), do: query
1199
1200   defp exclude_poll_votes(query, _) do
1201     if has_named_binding?(query, :object) do
1202       from([activity, object: o] in query,
1203         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1204       )
1205     else
1206       query
1207     end
1208   end
1209
1210   defp exclude_chat_messages(query, %{include_chat_messages: true}), do: query
1211
1212   defp exclude_chat_messages(query, _) do
1213     if has_named_binding?(query, :object) do
1214       from([activity, object: o] in query,
1215         where: fragment("not(?->>'type' = ?)", o.data, "ChatMessage")
1216       )
1217     else
1218       query
1219     end
1220   end
1221
1222   defp exclude_invisible_actors(query, %{invisible_actors: true}), do: query
1223
1224   defp exclude_invisible_actors(query, _opts) do
1225     invisible_ap_ids =
1226       User.Query.build(%{invisible: true, select: [:ap_id]})
1227       |> Repo.all()
1228       |> Enum.map(fn %{ap_id: ap_id} -> ap_id end)
1229
1230     from([activity] in query, where: activity.actor not in ^invisible_ap_ids)
1231   end
1232
1233   defp exclude_id(query, %{exclude_id: id}) when is_binary(id) do
1234     from(activity in query, where: activity.id != ^id)
1235   end
1236
1237   defp exclude_id(query, _), do: query
1238
1239   defp maybe_preload_objects(query, %{skip_preload: true}), do: query
1240
1241   defp maybe_preload_objects(query, _) do
1242     query
1243     |> Activity.with_preloaded_object()
1244   end
1245
1246   defp maybe_preload_bookmarks(query, %{skip_preload: true}), do: query
1247
1248   defp maybe_preload_bookmarks(query, opts) do
1249     query
1250     |> Activity.with_preloaded_bookmark(opts[:user])
1251   end
1252
1253   defp maybe_preload_report_notes(query, %{preload_report_notes: true}) do
1254     query
1255     |> Activity.with_preloaded_report_notes()
1256   end
1257
1258   defp maybe_preload_report_notes(query, _), do: query
1259
1260   defp maybe_set_thread_muted_field(query, %{skip_preload: true}), do: query
1261
1262   defp maybe_set_thread_muted_field(query, opts) do
1263     query
1264     |> Activity.with_set_thread_muted_field(opts[:muting_user] || opts[:user])
1265   end
1266
1267   defp maybe_order(query, %{order: :desc}) do
1268     query
1269     |> order_by(desc: :id)
1270   end
1271
1272   defp maybe_order(query, %{order: :asc}) do
1273     query
1274     |> order_by(asc: :id)
1275   end
1276
1277   defp maybe_order(query, _), do: query
1278
1279   defp normalize_fetch_activities_query_opts(opts) do
1280     Enum.reduce([:tag, :tag_all, :tag_reject], opts, fn key, opts ->
1281       case opts[key] do
1282         value when is_bitstring(value) ->
1283           Map.put(opts, key, Hashtag.normalize_name(value))
1284
1285         value when is_list(value) ->
1286           normalized_value =
1287             value
1288             |> Enum.map(&Hashtag.normalize_name/1)
1289             |> Enum.uniq()
1290
1291           Map.put(opts, key, normalized_value)
1292
1293         _ ->
1294           opts
1295       end
1296     end)
1297   end
1298
1299   defp fetch_activities_query_ap_ids_ops(opts) do
1300     source_user = opts[:muting_user]
1301     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1302
1303     ap_id_relationships =
1304       if opts[:blocking_user] && opts[:blocking_user] == source_user do
1305         [:block | ap_id_relationships]
1306       else
1307         ap_id_relationships
1308       end
1309
1310     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1311
1312     restrict_blocked_opts = Map.merge(%{blocked_users_ap_ids: preloaded_ap_ids[:block]}, opts)
1313     restrict_muted_opts = Map.merge(%{muted_users_ap_ids: preloaded_ap_ids[:mute]}, opts)
1314
1315     restrict_muted_reblogs_opts =
1316       Map.merge(%{reblog_muted_users_ap_ids: preloaded_ap_ids[:reblog_mute]}, opts)
1317
1318     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1319   end
1320
1321   def fetch_activities_query(recipients, opts \\ %{}) do
1322     opts = normalize_fetch_activities_query_opts(opts)
1323
1324     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1325       fetch_activities_query_ap_ids_ops(opts)
1326
1327     config = %{
1328       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1329     }
1330
1331     query =
1332       Activity
1333       |> maybe_preload_objects(opts)
1334       |> maybe_preload_bookmarks(opts)
1335       |> maybe_preload_report_notes(opts)
1336       |> maybe_set_thread_muted_field(opts)
1337       |> maybe_order(opts)
1338       |> restrict_recipients(recipients, opts[:user])
1339       |> restrict_replies(opts)
1340       |> restrict_since(opts)
1341       |> restrict_local(opts)
1342       |> restrict_remote(opts)
1343       |> restrict_actor(opts)
1344       |> restrict_type(opts)
1345       |> restrict_state(opts)
1346       |> restrict_favorited_by(opts)
1347       |> restrict_blocked(restrict_blocked_opts)
1348       |> restrict_blockers_visibility(opts)
1349       |> restrict_muted(restrict_muted_opts)
1350       |> restrict_filtered(opts)
1351       |> restrict_media(opts)
1352       |> restrict_visibility(opts)
1353       |> restrict_thread_visibility(opts, config)
1354       |> restrict_reblogs(opts)
1355       |> restrict_pinned(opts)
1356       |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1357       |> restrict_instance(opts)
1358       |> restrict_announce_object_actor(opts)
1359       |> restrict_filtered(opts)
1360       |> Activity.restrict_deactivated_users()
1361       |> exclude_poll_votes(opts)
1362       |> exclude_chat_messages(opts)
1363       |> exclude_invisible_actors(opts)
1364       |> exclude_visibility(opts)
1365
1366     if Config.feature_enabled?(:improved_hashtag_timeline) do
1367       query
1368       |> restrict_hashtag_any(opts)
1369       |> restrict_hashtag_all(opts)
1370       |> restrict_hashtag_reject_any(opts)
1371     else
1372       query
1373       |> restrict_embedded_tag_any(opts)
1374       |> restrict_embedded_tag_all(opts)
1375       |> restrict_embedded_tag_reject_any(opts)
1376     end
1377   end
1378
1379   @doc """
1380   Fetch favorites activities of user with order by sort adds to favorites
1381   """
1382   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1383   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1384     user.ap_id
1385     |> Activity.Queries.by_actor()
1386     |> Activity.Queries.by_type("Like")
1387     |> Activity.with_joined_object()
1388     |> Object.with_joined_activity()
1389     |> select([like, object, activity], %{activity | object: object, pagination_id: like.id})
1390     |> order_by([like, _, _], desc_nulls_last: like.id)
1391     |> Pagination.fetch_paginated(
1392       Map.merge(params, %{skip_order: true}),
1393       pagination
1394     )
1395   end
1396
1397   defp maybe_update_cc(activities, [_ | _] = list_memberships, %User{ap_id: user_ap_id}) do
1398     Enum.map(activities, fn
1399       %{data: %{"bcc" => [_ | _] = bcc}} = activity ->
1400         if Enum.any?(bcc, &(&1 in list_memberships)) do
1401           update_in(activity.data["cc"], &[user_ap_id | &1])
1402         else
1403           activity
1404         end
1405
1406       activity ->
1407         activity
1408     end)
1409   end
1410
1411   defp maybe_update_cc(activities, _, _), do: activities
1412
1413   defp fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1414     from(activity in query,
1415       where:
1416         fragment("? && ?", activity.recipients, ^recipients) or
1417           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1418              ^Constants.as_public() in activity.recipients)
1419     )
1420   end
1421
1422   def fetch_activities_bounded(
1423         recipients,
1424         recipients_with_public,
1425         opts \\ %{},
1426         pagination \\ :keyset
1427       ) do
1428     fetch_activities_query([], opts)
1429     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1430     |> Pagination.fetch_paginated(opts, pagination)
1431     |> Enum.reverse()
1432   end
1433
1434   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1435   def upload(file, opts \\ []) do
1436     with {:ok, data} <- Upload.store(file, opts) do
1437       obj_data = Maps.put_if_present(data, "actor", opts[:actor])
1438
1439       Repo.insert(%Object{data: obj_data})
1440     end
1441   end
1442
1443   @spec get_actor_url(any()) :: binary() | nil
1444   defp get_actor_url(url) when is_binary(url), do: url
1445   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1446
1447   defp get_actor_url(url) when is_list(url) do
1448     url
1449     |> List.first()
1450     |> get_actor_url()
1451   end
1452
1453   defp get_actor_url(_url), do: nil
1454
1455   defp normalize_image(%{"url" => url}) do
1456     %{
1457       "type" => "Image",
1458       "url" => [%{"href" => url}]
1459     }
1460   end
1461
1462   defp normalize_image(urls) when is_list(urls), do: urls |> List.first() |> normalize_image()
1463   defp normalize_image(_), do: nil
1464
1465   defp object_to_user_data(data) do
1466     fields =
1467       data
1468       |> Map.get("attachment", [])
1469       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1470       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1471
1472     emojis =
1473       data
1474       |> Map.get("tag", [])
1475       |> Enum.filter(fn
1476         %{"type" => "Emoji"} -> true
1477         _ -> false
1478       end)
1479       |> Map.new(fn %{"icon" => %{"url" => url}, "name" => name} ->
1480         {String.trim(name, ":"), url}
1481       end)
1482
1483     is_locked = data["manuallyApprovesFollowers"] || false
1484     capabilities = data["capabilities"] || %{}
1485     accepts_chat_messages = capabilities["acceptsChatMessages"]
1486     data = Transmogrifier.maybe_fix_user_object(data)
1487     is_discoverable = data["discoverable"] || false
1488     invisible = data["invisible"] || false
1489     actor_type = data["type"] || "Person"
1490
1491     featured_address = data["featured"]
1492     {:ok, pinned_objects} = fetch_and_prepare_featured_from_ap_id(featured_address)
1493
1494     public_key =
1495       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1496         data["publicKey"]["publicKeyPem"]
1497       else
1498         nil
1499       end
1500
1501     shared_inbox =
1502       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1503         data["endpoints"]["sharedInbox"]
1504       else
1505         nil
1506       end
1507
1508     user_data = %{
1509       ap_id: data["id"],
1510       uri: get_actor_url(data["url"]),
1511       ap_enabled: true,
1512       banner: normalize_image(data["image"]),
1513       fields: fields,
1514       emoji: emojis,
1515       is_locked: is_locked,
1516       is_discoverable: is_discoverable,
1517       invisible: invisible,
1518       avatar: normalize_image(data["icon"]),
1519       name: data["name"],
1520       follower_address: data["followers"],
1521       following_address: data["following"],
1522       featured_address: featured_address,
1523       bio: data["summary"] || "",
1524       actor_type: actor_type,
1525       also_known_as: Map.get(data, "alsoKnownAs", []),
1526       public_key: public_key,
1527       inbox: data["inbox"],
1528       shared_inbox: shared_inbox,
1529       accepts_chat_messages: accepts_chat_messages,
1530       pinned_objects: pinned_objects
1531     }
1532
1533     # nickname can be nil because of virtual actors
1534     if data["preferredUsername"] do
1535       Map.put(
1536         user_data,
1537         :nickname,
1538         "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1539       )
1540     else
1541       Map.put(user_data, :nickname, nil)
1542     end
1543   end
1544
1545   def fetch_follow_information_for_user(user) do
1546     with {:ok, following_data} <-
1547            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1548          {:ok, hide_follows} <- collection_private(following_data),
1549          {:ok, followers_data} <-
1550            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1551          {:ok, hide_followers} <- collection_private(followers_data) do
1552       {:ok,
1553        %{
1554          hide_follows: hide_follows,
1555          follower_count: normalize_counter(followers_data["totalItems"]),
1556          following_count: normalize_counter(following_data["totalItems"]),
1557          hide_followers: hide_followers
1558        }}
1559     else
1560       {:error, _} = e -> e
1561       e -> {:error, e}
1562     end
1563   end
1564
1565   defp normalize_counter(counter) when is_integer(counter), do: counter
1566   defp normalize_counter(_), do: 0
1567
1568   def maybe_update_follow_information(user_data) do
1569     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1570          {_, true} <- {:user_type_check, user_data[:type] in ["Person", "Service"]},
1571          {_, true} <-
1572            {:collections_available,
1573             !!(user_data[:following_address] && user_data[:follower_address])},
1574          {:ok, info} <-
1575            fetch_follow_information_for_user(user_data) do
1576       info = Map.merge(user_data[:info] || %{}, info)
1577
1578       user_data
1579       |> Map.put(:info, info)
1580     else
1581       {:user_type_check, false} ->
1582         user_data
1583
1584       {:collections_available, false} ->
1585         user_data
1586
1587       {:enabled, false} ->
1588         user_data
1589
1590       e ->
1591         Logger.error(
1592           "Follower/Following counter update for #{user_data.ap_id} failed.\n" <> inspect(e)
1593         )
1594
1595         user_data
1596     end
1597   end
1598
1599   defp collection_private(%{"first" => %{"type" => type}})
1600        when type in ["CollectionPage", "OrderedCollectionPage"],
1601        do: {:ok, false}
1602
1603   defp collection_private(%{"first" => first}) do
1604     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1605            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1606       {:ok, false}
1607     else
1608       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1609       {:error, _} = e -> e
1610       e -> {:error, e}
1611     end
1612   end
1613
1614   defp collection_private(_data), do: {:ok, true}
1615
1616   def user_data_from_user_object(data) do
1617     with {:ok, data} <- MRF.filter(data) do
1618       {:ok, object_to_user_data(data)}
1619     else
1620       e -> {:error, e}
1621     end
1622   end
1623
1624   def fetch_and_prepare_user_from_ap_id(ap_id) do
1625     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1626          {:ok, data} <- user_data_from_user_object(data) do
1627       {:ok, maybe_update_follow_information(data)}
1628     else
1629       # If this has been deleted, only log a debug and not an error
1630       {:error, "Object has been deleted" = e} ->
1631         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1632         {:error, e}
1633
1634       {:error, {:reject, reason} = e} ->
1635         Logger.info("Rejected user #{ap_id}: #{inspect(reason)}")
1636         {:error, e}
1637
1638       {:error, e} ->
1639         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1640         {:error, e}
1641     end
1642   end
1643
1644   def maybe_handle_clashing_nickname(data) do
1645     with nickname when is_binary(nickname) <- data[:nickname],
1646          %User{} = old_user <- User.get_by_nickname(nickname),
1647          {_, false} <- {:ap_id_comparison, data[:ap_id] == old_user.ap_id} do
1648       Logger.info(
1649         "Found an old user for #{nickname}, the old ap id is #{old_user.ap_id}, new one is #{data[:ap_id]}, renaming."
1650       )
1651
1652       old_user
1653       |> User.remote_user_changeset(%{nickname: "#{old_user.id}.#{old_user.nickname}"})
1654       |> User.update_and_set_cache()
1655     else
1656       {:ap_id_comparison, true} ->
1657         Logger.info(
1658           "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."
1659         )
1660
1661       _ ->
1662         nil
1663     end
1664   end
1665
1666   def pin_data_from_featured_collection(%{
1667         "type" => "OrderedCollection",
1668         "first" => first
1669       }) do
1670     with {:ok, page} <- Fetcher.fetch_and_contain_remote_object_from_id(first) do
1671       page
1672       |> Map.get("orderedItems")
1673       |> Map.new(fn %{"id" => object_ap_id} -> {object_ap_id, NaiveDateTime.utc_now()} end)
1674     else
1675       e ->
1676         Logger.error("Could not decode featured collection at fetch #{first}, #{inspect(e)}")
1677         {:ok, %{}}
1678     end
1679   end
1680
1681   def pin_data_from_featured_collection(
1682         %{
1683           "type" => type
1684         } = collection
1685       )
1686       when type in ["OrderedCollection", "Collection"] do
1687     {:ok, objects} = Collections.Fetcher.fetch_collection(collection)
1688     Map.new(objects, fn %{"id" => object_ap_id} -> {object_ap_id, NaiveDateTime.utc_now()} end)
1689   end
1690
1691   def fetch_and_prepare_featured_from_ap_id(nil) do
1692     {:ok, %{}}
1693   end
1694
1695   def fetch_and_prepare_featured_from_ap_id(ap_id) do
1696     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
1697       {:ok, pin_data_from_featured_collection(data)}
1698     else
1699       e ->
1700         Logger.error("Could not decode featured collection at fetch #{ap_id}, #{inspect(e)}")
1701         {:ok, %{}}
1702     end
1703   end
1704
1705   def pinned_fetch_task(nil), do: nil
1706
1707   def pinned_fetch_task(%{pinned_objects: pins}) do
1708     if Enum.all?(pins, fn {ap_id, _} ->
1709          Object.get_cached_by_ap_id(ap_id) ||
1710            match?({:ok, _object}, Fetcher.fetch_object_from_id(ap_id))
1711        end) do
1712       :ok
1713     else
1714       :error
1715     end
1716   end
1717
1718   def make_user_from_ap_id(ap_id) do
1719     user = User.get_cached_by_ap_id(ap_id)
1720
1721     if user && !User.ap_enabled?(user) do
1722       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1723     else
1724       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1725         {:ok, _pid} = Task.start(fn -> pinned_fetch_task(data) end)
1726
1727         if user do
1728           user
1729           |> User.remote_user_changeset(data)
1730           |> User.update_and_set_cache()
1731         else
1732           maybe_handle_clashing_nickname(data)
1733
1734           data
1735           |> User.remote_user_changeset()
1736           |> Repo.insert()
1737           |> User.set_cache()
1738         end
1739       end
1740     end
1741   end
1742
1743   def make_user_from_nickname(nickname) do
1744     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1745       make_user_from_ap_id(ap_id)
1746     else
1747       _e -> {:error, "No AP id in WebFinger"}
1748     end
1749   end
1750
1751   # filter out broken threads
1752   defp contain_broken_threads(%Activity{} = activity, %User{} = user) do
1753     entire_thread_visible_for_user?(activity, user)
1754   end
1755
1756   # do post-processing on a specific activity
1757   def contain_activity(%Activity{} = activity, %User{} = user) do
1758     contain_broken_threads(activity, user)
1759   end
1760
1761   def fetch_direct_messages_query do
1762     Activity
1763     |> restrict_type(%{type: "Create"})
1764     |> restrict_visibility(%{visibility: "direct"})
1765     |> order_by([activity], asc: activity.id)
1766   end
1767 end