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

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