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

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