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