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