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