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       User.all_superusers()
 675       |> Enum.filter(fn user -> not is_nil(user.email) end)
 676       |> Enum.each(fn superuser ->
 677         superuser
 678         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 679         |> Pleroma.Emails.Mailer.deliver_async()
 680       end)
 681
 682       {:ok, activity}
 683     end
 684   end
 685
 686   @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 687   def move(%User{} = origin, %User{} = target, local \\ true) do
 688     params = %{
 689       "type" => "Move",
 690       "actor" => origin.ap_id,
 691       "object" => origin.ap_id,
 692       "target" => target.ap_id
 693     }
 694
 695     with true <- origin.ap_id in target.also_known_as,
 696          {:ok, activity} <- insert(params, local) do
 697       maybe_federate(activity)
 698
 699       BackgroundWorker.enqueue("move_following", %{
 700         "origin_id" => origin.id,
 701         "target_id" => target.id
 702       })
 703
 704       {:ok, activity}
 705     else
 706       false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
 707       err -> err
 708     end
 709   end
 710
 711   defp fetch_activities_for_context_query(context, opts) do
 712     public = [Constants.as_public()]
 713
 714     recipients =
 715       if opts["user"],
 716         do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
 717         else: public
 718
 719     from(activity in Activity)
 720     |> maybe_preload_objects(opts)
 721     |> maybe_preload_bookmarks(opts)
 722     |> maybe_set_thread_muted_field(opts)
 723     |> restrict_blocked(opts)
 724     |> restrict_recipients(recipients, opts["user"])
 725     |> where(
 726       [activity],
 727       fragment(
 728         "?->>'type' = ? and ?->>'context' = ?",
 729         activity.data,
 730         "Create",
 731         activity.data,
 732         ^context
 733       )
 734     )
 735     |> exclude_poll_votes(opts)
 736     |> exclude_id(opts)
 737     |> order_by([activity], desc: activity.id)
 738   end
 739
 740   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 741   def fetch_activities_for_context(context, opts \\ %{}) do
 742     context
 743     |> fetch_activities_for_context_query(opts)
 744     |> Repo.all()
 745   end
 746
 747   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 748           FlakeId.Ecto.CompatType.t() | nil
 749   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 750     context
 751     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 752     |> limit(1)
 753     |> select([a], a.id)
 754     |> Repo.one()
 755   end
 756
 757   @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
 758   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 759     opts = Map.drop(opts, ["user"])
 760
 761     [Constants.as_public()]
 762     |> fetch_activities_query(opts)
 763     |> restrict_unlisted()
 764     |> Pagination.fetch_paginated(opts, pagination)
 765   end
 766
 767   @valid_visibilities ~w[direct unlisted public private]
 768
 769   defp restrict_visibility(query, %{visibility: visibility})
 770        when is_list(visibility) do
 771     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 772       query =
 773         from(
 774           a in query,
 775           where:
 776             fragment(
 777               "activity_visibility(?, ?, ?) = ANY (?)",
 778               a.actor,
 779               a.recipients,
 780               a.data,
 781               ^visibility
 782             )
 783         )
 784
 785       query
 786     else
 787       Logger.error("Could not restrict visibility to #{visibility}")
 788     end
 789   end
 790
 791   defp restrict_visibility(query, %{visibility: visibility})
 792        when visibility in @valid_visibilities do
 793     from(
 794       a in query,
 795       where:
 796         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 797     )
 798   end
 799
 800   defp restrict_visibility(_query, %{visibility: visibility})
 801        when visibility not in @valid_visibilities do
 802     Logger.error("Could not restrict visibility to #{visibility}")
 803   end
 804
 805   defp restrict_visibility(query, _visibility), do: query
 806
 807   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 808        when is_list(visibility) do
 809     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 810       from(
 811         a in query,
 812         where:
 813           not fragment(
 814             "activity_visibility(?, ?, ?) = ANY (?)",
 815             a.actor,
 816             a.recipients,
 817             a.data,
 818             ^visibility
 819           )
 820       )
 821     else
 822       Logger.error("Could not exclude visibility to #{visibility}")
 823       query
 824     end
 825   end
 826
 827   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 828        when visibility in @valid_visibilities do
 829     from(
 830       a in query,
 831       where:
 832         not fragment(
 833           "activity_visibility(?, ?, ?) = ?",
 834           a.actor,
 835           a.recipients,
 836           a.data,
 837           ^visibility
 838         )
 839     )
 840   end
 841
 842   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 843        when visibility not in @valid_visibilities do
 844     Logger.error("Could not exclude visibility to #{visibility}")
 845     query
 846   end
 847
 848   defp exclude_visibility(query, _visibility), do: query
 849
 850   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 851     do: query
 852
 853   defp restrict_thread_visibility(
 854          query,
 855          %{"user" => %User{skip_thread_containment: true}},
 856          _
 857        ),
 858        do: query
 859
 860   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 861     from(
 862       a in query,
 863       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 864     )
 865   end
 866
 867   defp restrict_thread_visibility(query, _, _), do: query
 868
 869   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 870     params =
 871       params
 872       |> Map.put("user", reading_user)
 873       |> Map.put("actor_id", user.ap_id)
 874
 875     recipients =
 876       user_activities_recipients(%{
 877         "godmode" => params["godmode"],
 878         "reading_user" => reading_user
 879       })
 880
 881     fetch_activities(recipients, params)
 882     |> Enum.reverse()
 883   end
 884
 885   def fetch_user_activities(user, reading_user, params \\ %{}) do
 886     params =
 887       params
 888       |> Map.put("type", ["Create", "Announce"])
 889       |> Map.put("user", reading_user)
 890       |> Map.put("actor_id", user.ap_id)
 891       |> Map.put("pinned_activity_ids", user.pinned_activities)
 892
 893     params =
 894       if User.blocks?(reading_user, user) do
 895         params
 896       else
 897         params
 898         |> Map.put("blocking_user", reading_user)
 899         |> Map.put("muting_user", reading_user)
 900       end
 901
 902     recipients =
 903       user_activities_recipients(%{
 904         "godmode" => params["godmode"],
 905         "reading_user" => reading_user
 906       })
 907
 908     fetch_activities(recipients, params)
 909     |> Enum.reverse()
 910   end
 911
 912   def fetch_statuses(reading_user, params) do
 913     params =
 914       params
 915       |> Map.put("type", ["Create", "Announce"])
 916
 917     recipients =
 918       user_activities_recipients(%{
 919         "godmode" => params["godmode"],
 920         "reading_user" => reading_user
 921       })
 922
 923     fetch_activities(recipients, params, :offset)
 924     |> Enum.reverse()
 925   end
 926
 927   defp user_activities_recipients(%{"godmode" => true}) do
 928     []
 929   end
 930
 931   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 932     if reading_user do
 933       [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 934     else
 935       [Constants.as_public()]
 936     end
 937   end
 938
 939   defp restrict_since(query, %{"since_id" => ""}), do: query
 940
 941   defp restrict_since(query, %{"since_id" => since_id}) do
 942     from(activity in query, where: activity.id > ^since_id)
 943   end
 944
 945   defp restrict_since(query, _), do: query
 946
 947   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 948     raise "Can't use the child object without preloading!"
 949   end
 950
 951   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 952        when is_list(tag_reject) and tag_reject != [] do
 953     from(
 954       [_activity, object] in query,
 955       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 956     )
 957   end
 958
 959   defp restrict_tag_reject(query, _), do: query
 960
 961   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 962     raise "Can't use the child object without preloading!"
 963   end
 964
 965   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 966        when is_list(tag_all) and tag_all != [] do
 967     from(
 968       [_activity, object] in query,
 969       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 970     )
 971   end
 972
 973   defp restrict_tag_all(query, _), do: query
 974
 975   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 976     raise "Can't use the child object without preloading!"
 977   end
 978
 979   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 980     from(
 981       [_activity, object] in query,
 982       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 983     )
 984   end
 985
 986   defp restrict_tag(query, %{"tag" => tag}) when is_binary(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, _), do: query
 994
 995   defp restrict_recipients(query, [], _user), do: query
 996
 997   defp restrict_recipients(query, recipients, nil) do
 998     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 999   end
1000
1001   defp restrict_recipients(query, recipients, user) do
1002     from(
1003       activity in query,
1004       where: fragment("? && ?", ^recipients, activity.recipients),
1005       or_where: activity.actor == ^user.ap_id
1006     )
1007   end
1008
1009   defp restrict_local(query, %{"local_only" => true}) do
1010     from(activity in query, where: activity.local == true)
1011   end
1012
1013   defp restrict_local(query, _), do: query
1014
1015   defp restrict_actor(query, %{"actor_id" => actor_id}) do
1016     from(activity in query, where: activity.actor == ^actor_id)
1017   end
1018
1019   defp restrict_actor(query, _), do: query
1020
1021   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
1022     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
1023   end
1024
1025   defp restrict_type(query, %{"type" => type}) do
1026     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
1027   end
1028
1029   defp restrict_type(query, _), do: query
1030
1031   defp restrict_state(query, %{"state" => state}) do
1032     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
1033   end
1034
1035   defp restrict_state(query, _), do: query
1036
1037   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
1038     from(
1039       [_activity, object] in query,
1040       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
1041     )
1042   end
1043
1044   defp restrict_favorited_by(query, _), do: query
1045
1046   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
1047     raise "Can't use the child object without preloading!"
1048   end
1049
1050   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
1051     from(
1052       [_activity, object] in query,
1053       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
1054     )
1055   end
1056
1057   defp restrict_media(query, _), do: query
1058
1059   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
1060     from(
1061       [_activity, object] in query,
1062       where: fragment("?->>'inReplyTo' is null", object.data)
1063     )
1064   end
1065
1066   defp restrict_replies(query, _), do: query
1067
1068   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
1069     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
1070   end
1071
1072   defp restrict_reblogs(query, _), do: query
1073
1074   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
1075
1076   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
1077     mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
1078
1079     query =
1080       from([activity] in query,
1081         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
1082         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
1083       )
1084
1085     unless opts["skip_preload"] do
1086       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
1087     else
1088       query
1089     end
1090   end
1091
1092   defp restrict_muted(query, _), do: query
1093
1094   defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
1095     blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
1096     domain_blocks = user.domain_blocks || []
1097
1098     following_ap_ids = User.get_friends_ap_ids(user)
1099
1100     query =
1101       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
1102
1103     from(
1104       [activity, object: o] in query,
1105       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
1106       where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
1107       where:
1108         fragment(
1109           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1110           activity.data,
1111           activity.data,
1112           ^blocked_ap_ids
1113         ),
1114       where:
1115         fragment(
1116           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
1117           activity.actor,
1118           ^domain_blocks,
1119           activity.actor,
1120           ^following_ap_ids
1121         ),
1122       where:
1123         fragment(
1124           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
1125           o.data,
1126           ^domain_blocks,
1127           o.data,
1128           ^following_ap_ids
1129         )
1130     )
1131   end
1132
1133   defp restrict_blocked(query, _), do: query
1134
1135   defp restrict_unlisted(query) do
1136     from(
1137       activity in query,
1138       where:
1139         fragment(
1140           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
1141           activity.data,
1142           ^[Constants.as_public()]
1143         )
1144     )
1145   end
1146
1147   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
1148     from(activity in query, where: activity.id in ^ids)
1149   end
1150
1151   defp restrict_pinned(query, _), do: query
1152
1153   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1154     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1155
1156     from(
1157       activity in query,
1158       where:
1159         fragment(
1160           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1161           activity.data,
1162           activity.actor,
1163           ^muted_reblogs
1164         )
1165     )
1166   end
1167
1168   defp restrict_muted_reblogs(query, _), do: query
1169
1170   defp restrict_instance(query, %{"instance" => instance}) do
1171     users =
1172       from(
1173         u in User,
1174         select: u.ap_id,
1175         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1176       )
1177       |> Repo.all()
1178
1179     from(activity in query, where: activity.actor in ^users)
1180   end
1181
1182   defp restrict_instance(query, _), do: query
1183
1184   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1185
1186   defp exclude_poll_votes(query, _) do
1187     if has_named_binding?(query, :object) do
1188       from([activity, object: o] in query,
1189         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1190       )
1191     else
1192       query
1193     end
1194   end
1195
1196   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1197     from(activity in query, where: activity.id != ^id)
1198   end
1199
1200   defp exclude_id(query, _), do: query
1201
1202   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1203
1204   defp maybe_preload_objects(query, _) do
1205     query
1206     |> Activity.with_preloaded_object()
1207   end
1208
1209   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1210
1211   defp maybe_preload_bookmarks(query, opts) do
1212     query
1213     |> Activity.with_preloaded_bookmark(opts["user"])
1214   end
1215
1216   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1217     query
1218     |> Activity.with_preloaded_report_notes()
1219   end
1220
1221   defp maybe_preload_report_notes(query, _), do: query
1222
1223   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1224
1225   defp maybe_set_thread_muted_field(query, opts) do
1226     query
1227     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1228   end
1229
1230   defp maybe_order(query, %{order: :desc}) do
1231     query
1232     |> order_by(desc: :id)
1233   end
1234
1235   defp maybe_order(query, %{order: :asc}) do
1236     query
1237     |> order_by(asc: :id)
1238   end
1239
1240   defp maybe_order(query, _), do: query
1241
1242   defp fetch_activities_query_ap_ids_ops(opts) do
1243     source_user = opts["muting_user"]
1244     ap_id_relations = if source_user, do: [:mute, :reblog_mute], else: []
1245
1246     ap_id_relations =
1247       ap_id_relations ++
1248         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1249           [:block]
1250         else
1251           []
1252         end
1253
1254     preloaded_ap_ids = User.outgoing_relations_ap_ids(source_user, ap_id_relations)
1255
1256     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1257     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1258
1259     restrict_muted_reblogs_opts =
1260       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1261
1262     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1263   end
1264
1265   def fetch_activities_query(recipients, opts \\ %{}) do
1266     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1267       fetch_activities_query_ap_ids_ops(opts)
1268
1269     config = %{
1270       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1271     }
1272
1273     Activity
1274     |> maybe_preload_objects(opts)
1275     |> maybe_preload_bookmarks(opts)
1276     |> maybe_preload_report_notes(opts)
1277     |> maybe_set_thread_muted_field(opts)
1278     |> maybe_order(opts)
1279     |> restrict_recipients(recipients, opts["user"])
1280     |> restrict_tag(opts)
1281     |> restrict_tag_reject(opts)
1282     |> restrict_tag_all(opts)
1283     |> restrict_since(opts)
1284     |> restrict_local(opts)
1285     |> restrict_actor(opts)
1286     |> restrict_type(opts)
1287     |> restrict_state(opts)
1288     |> restrict_favorited_by(opts)
1289     |> restrict_blocked(restrict_blocked_opts)
1290     |> restrict_muted(restrict_muted_opts)
1291     |> restrict_media(opts)
1292     |> restrict_visibility(opts)
1293     |> restrict_thread_visibility(opts, config)
1294     |> restrict_replies(opts)
1295     |> restrict_reblogs(opts)
1296     |> restrict_pinned(opts)
1297     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1298     |> restrict_instance(opts)
1299     |> Activity.restrict_deactivated_users()
1300     |> exclude_poll_votes(opts)
1301     |> exclude_visibility(opts)
1302   end
1303
1304   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1305     list_memberships = Pleroma.List.memberships(opts["user"])
1306
1307     fetch_activities_query(recipients ++ list_memberships, opts)
1308     |> Pagination.fetch_paginated(opts, pagination)
1309     |> Enum.reverse()
1310     |> maybe_update_cc(list_memberships, opts["user"])
1311   end
1312
1313   @doc """
1314   Fetch favorites activities of user with order by sort adds to favorites
1315   """
1316   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1317   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1318     user.ap_id
1319     |> Activity.Queries.by_actor()
1320     |> Activity.Queries.by_type("Like")
1321     |> Activity.with_joined_object()
1322     |> Object.with_joined_activity()
1323     |> select([_like, object, activity], %{activity | object: object})
1324     |> order_by([like, _, _], desc: like.id)
1325     |> Pagination.fetch_paginated(
1326       Map.merge(params, %{"skip_order" => true}),
1327       pagination,
1328       :object_activity
1329     )
1330   end
1331
1332   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1333        when is_list(list_memberships) and length(list_memberships) > 0 do
1334     Enum.map(activities, fn
1335       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1336         if Enum.any?(bcc, &(&1 in list_memberships)) do
1337           update_in(activity.data["cc"], &[user_ap_id | &1])
1338         else
1339           activity
1340         end
1341
1342       activity ->
1343         activity
1344     end)
1345   end
1346
1347   defp maybe_update_cc(activities, _, _), do: activities
1348
1349   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1350     from(activity in query,
1351       where:
1352         fragment("? && ?", activity.recipients, ^recipients) or
1353           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1354              ^Constants.as_public() in activity.recipients)
1355     )
1356   end
1357
1358   def fetch_activities_bounded(
1359         recipients,
1360         recipients_with_public,
1361         opts \\ %{},
1362         pagination \\ :keyset
1363       ) do
1364     fetch_activities_query([], opts)
1365     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1366     |> Pagination.fetch_paginated(opts, pagination)
1367     |> Enum.reverse()
1368   end
1369
1370   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1371   def upload(file, opts \\ []) do
1372     with {:ok, data} <- Upload.store(file, opts) do
1373       obj_data =
1374         if opts[:actor] do
1375           Map.put(data, "actor", opts[:actor])
1376         else
1377           data
1378         end
1379
1380       Repo.insert(%Object{data: obj_data})
1381     end
1382   end
1383
1384   defp object_to_user_data(data) do
1385     avatar =
1386       data["icon"]["url"] &&
1387         %{
1388           "type" => "Image",
1389           "url" => [%{"href" => data["icon"]["url"]}]
1390         }
1391
1392     banner =
1393       data["image"]["url"] &&
1394         %{
1395           "type" => "Image",
1396           "url" => [%{"href" => data["image"]["url"]}]
1397         }
1398
1399     fields =
1400       data
1401       |> Map.get("attachment", [])
1402       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1403       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1404
1405     locked = data["manuallyApprovesFollowers"] || false
1406     data = Transmogrifier.maybe_fix_user_object(data)
1407     discoverable = data["discoverable"] || false
1408     invisible = data["invisible"] || false
1409     actor_type = data["type"] || "Person"
1410
1411     user_data = %{
1412       ap_id: data["id"],
1413       ap_enabled: true,
1414       source_data: data,
1415       banner: banner,
1416       fields: fields,
1417       locked: locked,
1418       discoverable: discoverable,
1419       invisible: invisible,
1420       avatar: avatar,
1421       name: data["name"],
1422       follower_address: data["followers"],
1423       following_address: data["following"],
1424       bio: data["summary"],
1425       actor_type: actor_type,
1426       also_known_as: Map.get(data, "alsoKnownAs", [])
1427     }
1428
1429     # nickname can be nil because of virtual actors
1430     user_data =
1431       if data["preferredUsername"] do
1432         Map.put(
1433           user_data,
1434           :nickname,
1435           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1436         )
1437       else
1438         Map.put(user_data, :nickname, nil)
1439       end
1440
1441     {:ok, user_data}
1442   end
1443
1444   def fetch_follow_information_for_user(user) do
1445     with {:ok, following_data} <-
1446            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1447          {:ok, hide_follows} <- collection_private(following_data),
1448          {:ok, followers_data} <-
1449            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1450          {:ok, hide_followers} <- collection_private(followers_data) do
1451       {:ok,
1452        %{
1453          hide_follows: hide_follows,
1454          follower_count: normalize_counter(followers_data["totalItems"]),
1455          following_count: normalize_counter(following_data["totalItems"]),
1456          hide_followers: hide_followers
1457        }}
1458     else
1459       {:error, _} = e -> e
1460       e -> {:error, e}
1461     end
1462   end
1463
1464   defp normalize_counter(counter) when is_integer(counter), do: counter
1465   defp normalize_counter(_), do: 0
1466
1467   defp maybe_update_follow_information(data) do
1468     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1469          {:ok, info} <- fetch_follow_information_for_user(data) do
1470       info = Map.merge(data[:info] || %{}, info)
1471       Map.put(data, :info, info)
1472     else
1473       {:enabled, false} ->
1474         data
1475
1476       e ->
1477         Logger.error(
1478           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1479         )
1480
1481         data
1482     end
1483   end
1484
1485   defp collection_private(%{"first" => %{"type" => type}})
1486        when type in ["CollectionPage", "OrderedCollectionPage"],
1487        do: {:ok, false}
1488
1489   defp collection_private(%{"first" => first}) do
1490     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1491            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1492       {:ok, false}
1493     else
1494       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1495       {:error, _} = e -> e
1496       e -> {:error, e}
1497     end
1498   end
1499
1500   defp collection_private(_data), do: {:ok, true}
1501
1502   def user_data_from_user_object(data) do
1503     with {:ok, data} <- MRF.filter(data),
1504          {:ok, data} <- object_to_user_data(data) do
1505       {:ok, data}
1506     else
1507       e -> {:error, e}
1508     end
1509   end
1510
1511   def fetch_and_prepare_user_from_ap_id(ap_id) do
1512     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1513          {:ok, data} <- user_data_from_user_object(data),
1514          data <- maybe_update_follow_information(data) do
1515       {:ok, data}
1516     else
1517       {:error, "Object has been deleted"} = e ->
1518         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1519         {:error, e}
1520
1521       e ->
1522         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1523         {:error, e}
1524     end
1525   end
1526
1527   def make_user_from_ap_id(ap_id) do
1528     if _user = User.get_cached_by_ap_id(ap_id) do
1529       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1530     else
1531       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1532         User.insert_or_update_user(data)
1533       else
1534         e -> {:error, e}
1535       end
1536     end
1537   end
1538
1539   def make_user_from_nickname(nickname) do
1540     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1541       make_user_from_ap_id(ap_id)
1542     else
1543       _e -> {:error, "No AP id in WebFinger"}
1544     end
1545   end
1546
1547   # filter out broken threads
1548   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1549     entire_thread_visible_for_user?(activity, user)
1550   end
1551
1552   # do post-processing on a specific activity
1553   def contain_activity(%Activity{} = activity, %User{} = user) do
1554     contain_broken_threads(activity, user)
1555   end
1556
1557   def fetch_direct_messages_query do
1558     Activity
1559     |> restrict_type(%{"type" => "Create"})
1560     |> restrict_visibility(%{visibility: "direct"})
1561     |> order_by([activity], asc: activity.id)
1562   end
1563 end