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