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