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 [nil | @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 in [true, "true", "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 in [true, "true", "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 in [true, "true", "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   # TODO: when all endpoints migrated to OpenAPI compare `pinned` with `true` (boolean) only,
1168   # the same for `restrict_media/2`, `restrict_replies/2`, 'restrict_reblogs/2'
1169   # and `restrict_muted/2`
1170
1171   defp restrict_pinned(query, %{"pinned" => pinned, "pinned_activity_ids" => ids})
1172        when pinned in [true, "true", "1"] do
1173     from(activity in query, where: activity.id in ^ids)
1174   end
1175
1176   defp restrict_pinned(query, _), do: query
1177
1178   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1179     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1180
1181     from(
1182       activity in query,
1183       where:
1184         fragment(
1185           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1186           activity.data,
1187           activity.actor,
1188           ^muted_reblogs
1189         )
1190     )
1191   end
1192
1193   defp restrict_muted_reblogs(query, _), do: query
1194
1195   defp restrict_instance(query, %{"instance" => instance}) do
1196     users =
1197       from(
1198         u in User,
1199         select: u.ap_id,
1200         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1201       )
1202       |> Repo.all()
1203
1204     from(activity in query, where: activity.actor in ^users)
1205   end
1206
1207   defp restrict_instance(query, _), do: query
1208
1209   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1210
1211   defp exclude_poll_votes(query, _) do
1212     if has_named_binding?(query, :object) do
1213       from([activity, object: o] in query,
1214         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1215       )
1216     else
1217       query
1218     end
1219   end
1220
1221   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1222     from(activity in query, where: activity.id != ^id)
1223   end
1224
1225   defp exclude_id(query, _), do: query
1226
1227   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1228
1229   defp maybe_preload_objects(query, _) do
1230     query
1231     |> Activity.with_preloaded_object()
1232   end
1233
1234   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1235
1236   defp maybe_preload_bookmarks(query, opts) do
1237     query
1238     |> Activity.with_preloaded_bookmark(opts["user"])
1239   end
1240
1241   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1242     query
1243     |> Activity.with_preloaded_report_notes()
1244   end
1245
1246   defp maybe_preload_report_notes(query, _), do: query
1247
1248   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1249
1250   defp maybe_set_thread_muted_field(query, opts) do
1251     query
1252     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1253   end
1254
1255   defp maybe_order(query, %{order: :desc}) do
1256     query
1257     |> order_by(desc: :id)
1258   end
1259
1260   defp maybe_order(query, %{order: :asc}) do
1261     query
1262     |> order_by(asc: :id)
1263   end
1264
1265   defp maybe_order(query, _), do: query
1266
1267   defp fetch_activities_query_ap_ids_ops(opts) do
1268     source_user = opts["muting_user"]
1269     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1270
1271     ap_id_relationships =
1272       ap_id_relationships ++
1273         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1274           [:block]
1275         else
1276           []
1277         end
1278
1279     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1280
1281     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1282     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1283
1284     restrict_muted_reblogs_opts =
1285       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1286
1287     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1288   end
1289
1290   def fetch_activities_query(recipients, opts \\ %{}) do
1291     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1292       fetch_activities_query_ap_ids_ops(opts)
1293
1294     config = %{
1295       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1296     }
1297
1298     Activity
1299     |> maybe_preload_objects(opts)
1300     |> maybe_preload_bookmarks(opts)
1301     |> maybe_preload_report_notes(opts)
1302     |> maybe_set_thread_muted_field(opts)
1303     |> maybe_order(opts)
1304     |> restrict_recipients(recipients, opts["user"])
1305     |> restrict_replies(opts)
1306     |> restrict_tag(opts)
1307     |> restrict_tag_reject(opts)
1308     |> restrict_tag_all(opts)
1309     |> restrict_since(opts)
1310     |> restrict_local(opts)
1311     |> restrict_actor(opts)
1312     |> restrict_type(opts)
1313     |> restrict_state(opts)
1314     |> restrict_favorited_by(opts)
1315     |> restrict_blocked(restrict_blocked_opts)
1316     |> restrict_muted(restrict_muted_opts)
1317     |> restrict_media(opts)
1318     |> restrict_visibility(opts)
1319     |> restrict_thread_visibility(opts, config)
1320     |> restrict_reblogs(opts)
1321     |> restrict_pinned(opts)
1322     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1323     |> restrict_instance(opts)
1324     |> Activity.restrict_deactivated_users()
1325     |> exclude_poll_votes(opts)
1326     |> exclude_visibility(opts)
1327   end
1328
1329   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1330     list_memberships = Pleroma.List.memberships(opts["user"])
1331
1332     fetch_activities_query(recipients ++ list_memberships, opts)
1333     |> Pagination.fetch_paginated(opts, pagination)
1334     |> Enum.reverse()
1335     |> maybe_update_cc(list_memberships, opts["user"])
1336   end
1337
1338   @doc """
1339   Fetch favorites activities of user with order by sort adds to favorites
1340   """
1341   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1342   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1343     user.ap_id
1344     |> Activity.Queries.by_actor()
1345     |> Activity.Queries.by_type("Like")
1346     |> Activity.with_joined_object()
1347     |> Object.with_joined_activity()
1348     |> select([_like, object, activity], %{activity | object: object})
1349     |> order_by([like, _, _], desc: like.id)
1350     |> Pagination.fetch_paginated(
1351       Map.merge(params, %{"skip_order" => true}),
1352       pagination,
1353       :object_activity
1354     )
1355   end
1356
1357   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1358        when is_list(list_memberships) and length(list_memberships) > 0 do
1359     Enum.map(activities, fn
1360       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1361         if Enum.any?(bcc, &(&1 in list_memberships)) do
1362           update_in(activity.data["cc"], &[user_ap_id | &1])
1363         else
1364           activity
1365         end
1366
1367       activity ->
1368         activity
1369     end)
1370   end
1371
1372   defp maybe_update_cc(activities, _, _), do: activities
1373
1374   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1375     from(activity in query,
1376       where:
1377         fragment("? && ?", activity.recipients, ^recipients) or
1378           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1379              ^Constants.as_public() in activity.recipients)
1380     )
1381   end
1382
1383   def fetch_activities_bounded(
1384         recipients,
1385         recipients_with_public,
1386         opts \\ %{},
1387         pagination \\ :keyset
1388       ) do
1389     fetch_activities_query([], opts)
1390     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1391     |> Pagination.fetch_paginated(opts, pagination)
1392     |> Enum.reverse()
1393   end
1394
1395   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1396   def upload(file, opts \\ []) do
1397     with {:ok, data} <- Upload.store(file, opts) do
1398       obj_data =
1399         if opts[:actor] do
1400           Map.put(data, "actor", opts[:actor])
1401         else
1402           data
1403         end
1404
1405       Repo.insert(%Object{data: obj_data})
1406     end
1407   end
1408
1409   @spec get_actor_url(any()) :: binary() | nil
1410   defp get_actor_url(url) when is_binary(url), do: url
1411   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1412
1413   defp get_actor_url(url) when is_list(url) do
1414     url
1415     |> List.first()
1416     |> get_actor_url()
1417   end
1418
1419   defp get_actor_url(_url), do: nil
1420
1421   defp object_to_user_data(data) do
1422     avatar =
1423       data["icon"]["url"] &&
1424         %{
1425           "type" => "Image",
1426           "url" => [%{"href" => data["icon"]["url"]}]
1427         }
1428
1429     banner =
1430       data["image"]["url"] &&
1431         %{
1432           "type" => "Image",
1433           "url" => [%{"href" => data["image"]["url"]}]
1434         }
1435
1436     fields =
1437       data
1438       |> Map.get("attachment", [])
1439       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1440       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1441
1442     emojis =
1443       data
1444       |> Map.get("tag", [])
1445       |> Enum.filter(fn
1446         %{"type" => "Emoji"} -> true
1447         _ -> false
1448       end)
1449       |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1450         Map.put(acc, String.trim(name, ":"), url)
1451       end)
1452
1453     locked = data["manuallyApprovesFollowers"] || false
1454     data = Transmogrifier.maybe_fix_user_object(data)
1455     discoverable = data["discoverable"] || false
1456     invisible = data["invisible"] || false
1457     actor_type = data["type"] || "Person"
1458
1459     public_key =
1460       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1461         data["publicKey"]["publicKeyPem"]
1462       else
1463         nil
1464       end
1465
1466     shared_inbox =
1467       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1468         data["endpoints"]["sharedInbox"]
1469       else
1470         nil
1471       end
1472
1473     user_data = %{
1474       ap_id: data["id"],
1475       uri: get_actor_url(data["url"]),
1476       ap_enabled: true,
1477       banner: banner,
1478       fields: fields,
1479       emoji: emojis,
1480       locked: locked,
1481       discoverable: discoverable,
1482       invisible: invisible,
1483       avatar: avatar,
1484       name: data["name"],
1485       follower_address: data["followers"],
1486       following_address: data["following"],
1487       bio: data["summary"],
1488       actor_type: actor_type,
1489       also_known_as: Map.get(data, "alsoKnownAs", []),
1490       public_key: public_key,
1491       inbox: data["inbox"],
1492       shared_inbox: shared_inbox
1493     }
1494
1495     # nickname can be nil because of virtual actors
1496     user_data =
1497       if data["preferredUsername"] do
1498         Map.put(
1499           user_data,
1500           :nickname,
1501           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1502         )
1503       else
1504         Map.put(user_data, :nickname, nil)
1505       end
1506
1507     {:ok, user_data}
1508   end
1509
1510   def fetch_follow_information_for_user(user) do
1511     with {:ok, following_data} <-
1512            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1513          {:ok, hide_follows} <- collection_private(following_data),
1514          {:ok, followers_data} <-
1515            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1516          {:ok, hide_followers} <- collection_private(followers_data) do
1517       {:ok,
1518        %{
1519          hide_follows: hide_follows,
1520          follower_count: normalize_counter(followers_data["totalItems"]),
1521          following_count: normalize_counter(following_data["totalItems"]),
1522          hide_followers: hide_followers
1523        }}
1524     else
1525       {:error, _} = e -> e
1526       e -> {:error, e}
1527     end
1528   end
1529
1530   defp normalize_counter(counter) when is_integer(counter), do: counter
1531   defp normalize_counter(_), do: 0
1532
1533   defp maybe_update_follow_information(data) do
1534     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1535          {:ok, info} <- fetch_follow_information_for_user(data) do
1536       info = Map.merge(data[:info] || %{}, info)
1537       Map.put(data, :info, info)
1538     else
1539       {:enabled, false} ->
1540         data
1541
1542       e ->
1543         Logger.error(
1544           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1545         )
1546
1547         data
1548     end
1549   end
1550
1551   defp collection_private(%{"first" => %{"type" => type}})
1552        when type in ["CollectionPage", "OrderedCollectionPage"],
1553        do: {:ok, false}
1554
1555   defp collection_private(%{"first" => first}) do
1556     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1557            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1558       {:ok, false}
1559     else
1560       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1561       {:error, _} = e -> e
1562       e -> {:error, e}
1563     end
1564   end
1565
1566   defp collection_private(_data), do: {:ok, true}
1567
1568   def user_data_from_user_object(data) do
1569     with {:ok, data} <- MRF.filter(data),
1570          {:ok, data} <- object_to_user_data(data) do
1571       {:ok, data}
1572     else
1573       e -> {:error, e}
1574     end
1575   end
1576
1577   def fetch_and_prepare_user_from_ap_id(ap_id) do
1578     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1579          {:ok, data} <- user_data_from_user_object(data),
1580          data <- maybe_update_follow_information(data) do
1581       {:ok, data}
1582     else
1583       {:error, "Object has been deleted"} = e ->
1584         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1585         {:error, e}
1586
1587       e ->
1588         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1589         {:error, e}
1590     end
1591   end
1592
1593   def make_user_from_ap_id(ap_id) do
1594     user = User.get_cached_by_ap_id(ap_id)
1595
1596     if user && !User.ap_enabled?(user) do
1597       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1598     else
1599       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1600         if user do
1601           user
1602           |> User.remote_user_changeset(data)
1603           |> User.update_and_set_cache()
1604         else
1605           data
1606           |> User.remote_user_changeset()
1607           |> Repo.insert()
1608           |> User.set_cache()
1609         end
1610       else
1611         e -> {:error, e}
1612       end
1613     end
1614   end
1615
1616   def make_user_from_nickname(nickname) do
1617     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1618       make_user_from_ap_id(ap_id)
1619     else
1620       _e -> {:error, "No AP id in WebFinger"}
1621     end
1622   end
1623
1624   # filter out broken threads
1625   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1626     entire_thread_visible_for_user?(activity, user)
1627   end
1628
1629   # do post-processing on a specific activity
1630   def contain_activity(%Activity{} = activity, %User{} = user) do
1631     contain_broken_threads(activity, user)
1632   end
1633
1634   def fetch_direct_messages_query do
1635     Activity
1636     |> restrict_type(%{"type" => "Create"})
1637     |> restrict_visibility(%{visibility: "direct"})
1638     |> order_by([activity], asc: activity.id)
1639   end
1640 end