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