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

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