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

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