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

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