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

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