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