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