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