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