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   # TODO: when all endpoints migrated to OpenAPI compare `pinned` with `true` (boolean) only,
1161   # the same for `restrict_media/2`, `restrict_replies/2`, 'restrict_reblogs/2'
1162   # and `restrict_muted/2`
1163
1164   defp restrict_pinned(query, %{"pinned" => pinned, "pinned_activity_ids" => ids})
1165        when pinned in [true, "true", "1"] do
1166     from(activity in query, where: activity.id in ^ids)
1167   end
1168
1169   defp restrict_pinned(query, _), do: query
1170
1171   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1172     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1173
1174     from(
1175       activity in query,
1176       where:
1177         fragment(
1178           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1179           activity.data,
1180           activity.actor,
1181           ^muted_reblogs
1182         )
1183     )
1184   end
1185
1186   defp restrict_muted_reblogs(query, _), do: query
1187
1188   defp restrict_instance(query, %{"instance" => instance}) do
1189     users =
1190       from(
1191         u in User,
1192         select: u.ap_id,
1193         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1194       )
1195       |> Repo.all()
1196
1197     from(activity in query, where: activity.actor in ^users)
1198   end
1199
1200   defp restrict_instance(query, _), do: query
1201
1202   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1203
1204   defp exclude_poll_votes(query, _) do
1205     if has_named_binding?(query, :object) do
1206       from([activity, object: o] in query,
1207         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1208       )
1209     else
1210       query
1211     end
1212   end
1213
1214   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1215     from(activity in query, where: activity.id != ^id)
1216   end
1217
1218   defp exclude_id(query, _), do: query
1219
1220   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1221
1222   defp maybe_preload_objects(query, _) do
1223     query
1224     |> Activity.with_preloaded_object()
1225   end
1226
1227   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1228
1229   defp maybe_preload_bookmarks(query, opts) do
1230     query
1231     |> Activity.with_preloaded_bookmark(opts["user"])
1232   end
1233
1234   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1235     query
1236     |> Activity.with_preloaded_report_notes()
1237   end
1238
1239   defp maybe_preload_report_notes(query, _), do: query
1240
1241   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1242
1243   defp maybe_set_thread_muted_field(query, opts) do
1244     query
1245     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1246   end
1247
1248   defp maybe_order(query, %{order: :desc}) do
1249     query
1250     |> order_by(desc: :id)
1251   end
1252
1253   defp maybe_order(query, %{order: :asc}) do
1254     query
1255     |> order_by(asc: :id)
1256   end
1257
1258   defp maybe_order(query, _), do: query
1259
1260   defp fetch_activities_query_ap_ids_ops(opts) do
1261     source_user = opts["muting_user"]
1262     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1263
1264     ap_id_relationships =
1265       ap_id_relationships ++
1266         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1267           [:block]
1268         else
1269           []
1270         end
1271
1272     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1273
1274     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1275     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1276
1277     restrict_muted_reblogs_opts =
1278       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1279
1280     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1281   end
1282
1283   def fetch_activities_query(recipients, opts \\ %{}) do
1284     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1285       fetch_activities_query_ap_ids_ops(opts)
1286
1287     config = %{
1288       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1289     }
1290
1291     Activity
1292     |> maybe_preload_objects(opts)
1293     |> maybe_preload_bookmarks(opts)
1294     |> maybe_preload_report_notes(opts)
1295     |> maybe_set_thread_muted_field(opts)
1296     |> maybe_order(opts)
1297     |> restrict_recipients(recipients, opts["user"])
1298     |> restrict_tag(opts)
1299     |> restrict_tag_reject(opts)
1300     |> restrict_tag_all(opts)
1301     |> restrict_since(opts)
1302     |> restrict_local(opts)
1303     |> restrict_actor(opts)
1304     |> restrict_type(opts)
1305     |> restrict_state(opts)
1306     |> restrict_favorited_by(opts)
1307     |> restrict_blocked(restrict_blocked_opts)
1308     |> restrict_muted(restrict_muted_opts)
1309     |> restrict_media(opts)
1310     |> restrict_visibility(opts)
1311     |> restrict_thread_visibility(opts, config)
1312     |> restrict_replies(opts)
1313     |> restrict_reblogs(opts)
1314     |> restrict_pinned(opts)
1315     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1316     |> restrict_instance(opts)
1317     |> Activity.restrict_deactivated_users()
1318     |> exclude_poll_votes(opts)
1319     |> exclude_visibility(opts)
1320   end
1321
1322   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1323     list_memberships = Pleroma.List.memberships(opts["user"])
1324
1325     fetch_activities_query(recipients ++ list_memberships, opts)
1326     |> Pagination.fetch_paginated(opts, pagination)
1327     |> Enum.reverse()
1328     |> maybe_update_cc(list_memberships, opts["user"])
1329   end
1330
1331   @doc """
1332   Fetch favorites activities of user with order by sort adds to favorites
1333   """
1334   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1335   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1336     user.ap_id
1337     |> Activity.Queries.by_actor()
1338     |> Activity.Queries.by_type("Like")
1339     |> Activity.with_joined_object()
1340     |> Object.with_joined_activity()
1341     |> select([_like, object, activity], %{activity | object: object})
1342     |> order_by([like, _, _], desc: like.id)
1343     |> Pagination.fetch_paginated(
1344       Map.merge(params, %{"skip_order" => true}),
1345       pagination,
1346       :object_activity
1347     )
1348   end
1349
1350   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1351        when is_list(list_memberships) and length(list_memberships) > 0 do
1352     Enum.map(activities, fn
1353       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1354         if Enum.any?(bcc, &(&1 in list_memberships)) do
1355           update_in(activity.data["cc"], &[user_ap_id | &1])
1356         else
1357           activity
1358         end
1359
1360       activity ->
1361         activity
1362     end)
1363   end
1364
1365   defp maybe_update_cc(activities, _, _), do: activities
1366
1367   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1368     from(activity in query,
1369       where:
1370         fragment("? && ?", activity.recipients, ^recipients) or
1371           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1372              ^Constants.as_public() in activity.recipients)
1373     )
1374   end
1375
1376   def fetch_activities_bounded(
1377         recipients,
1378         recipients_with_public,
1379         opts \\ %{},
1380         pagination \\ :keyset
1381       ) do
1382     fetch_activities_query([], opts)
1383     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1384     |> Pagination.fetch_paginated(opts, pagination)
1385     |> Enum.reverse()
1386   end
1387
1388   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1389   def upload(file, opts \\ []) do
1390     with {:ok, data} <- Upload.store(file, opts) do
1391       obj_data =
1392         if opts[:actor] do
1393           Map.put(data, "actor", opts[:actor])
1394         else
1395           data
1396         end
1397
1398       Repo.insert(%Object{data: obj_data})
1399     end
1400   end
1401
1402   @spec get_actor_url(any()) :: binary() | nil
1403   defp get_actor_url(url) when is_binary(url), do: url
1404   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1405
1406   defp get_actor_url(url) when is_list(url) do
1407     url
1408     |> List.first()
1409     |> get_actor_url()
1410   end
1411
1412   defp get_actor_url(_url), do: nil
1413
1414   defp object_to_user_data(data) do
1415     avatar =
1416       data["icon"]["url"] &&
1417         %{
1418           "type" => "Image",
1419           "url" => [%{"href" => data["icon"]["url"]}]
1420         }
1421
1422     banner =
1423       data["image"]["url"] &&
1424         %{
1425           "type" => "Image",
1426           "url" => [%{"href" => data["image"]["url"]}]
1427         }
1428
1429     fields =
1430       data
1431       |> Map.get("attachment", [])
1432       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1433       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1434
1435     emojis =
1436       data
1437       |> Map.get("tag", [])
1438       |> Enum.filter(fn
1439         %{"type" => "Emoji"} -> true
1440         _ -> false
1441       end)
1442       |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1443         Map.put(acc, String.trim(name, ":"), url)
1444       end)
1445
1446     locked = data["manuallyApprovesFollowers"] || false
1447     data = Transmogrifier.maybe_fix_user_object(data)
1448     discoverable = data["discoverable"] || false
1449     invisible = data["invisible"] || false
1450     actor_type = data["type"] || "Person"
1451
1452     public_key =
1453       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1454         data["publicKey"]["publicKeyPem"]
1455       else
1456         nil
1457       end
1458
1459     shared_inbox =
1460       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1461         data["endpoints"]["sharedInbox"]
1462       else
1463         nil
1464       end
1465
1466     user_data = %{
1467       ap_id: data["id"],
1468       uri: get_actor_url(data["url"]),
1469       ap_enabled: true,
1470       banner: banner,
1471       fields: fields,
1472       emoji: emojis,
1473       locked: locked,
1474       discoverable: discoverable,
1475       invisible: invisible,
1476       avatar: avatar,
1477       name: data["name"],
1478       follower_address: data["followers"],
1479       following_address: data["following"],
1480       bio: data["summary"],
1481       actor_type: actor_type,
1482       also_known_as: Map.get(data, "alsoKnownAs", []),
1483       public_key: public_key,
1484       inbox: data["inbox"],
1485       shared_inbox: shared_inbox
1486     }
1487
1488     # nickname can be nil because of virtual actors
1489     user_data =
1490       if data["preferredUsername"] do
1491         Map.put(
1492           user_data,
1493           :nickname,
1494           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1495         )
1496       else
1497         Map.put(user_data, :nickname, nil)
1498       end
1499
1500     {:ok, user_data}
1501   end
1502
1503   def fetch_follow_information_for_user(user) do
1504     with {:ok, following_data} <-
1505            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1506          {:ok, hide_follows} <- collection_private(following_data),
1507          {:ok, followers_data} <-
1508            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1509          {:ok, hide_followers} <- collection_private(followers_data) do
1510       {:ok,
1511        %{
1512          hide_follows: hide_follows,
1513          follower_count: normalize_counter(followers_data["totalItems"]),
1514          following_count: normalize_counter(following_data["totalItems"]),
1515          hide_followers: hide_followers
1516        }}
1517     else
1518       {:error, _} = e -> e
1519       e -> {:error, e}
1520     end
1521   end
1522
1523   defp normalize_counter(counter) when is_integer(counter), do: counter
1524   defp normalize_counter(_), do: 0
1525
1526   defp maybe_update_follow_information(data) do
1527     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1528          {:ok, info} <- fetch_follow_information_for_user(data) do
1529       info = Map.merge(data[:info] || %{}, info)
1530       Map.put(data, :info, info)
1531     else
1532       {:enabled, false} ->
1533         data
1534
1535       e ->
1536         Logger.error(
1537           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1538         )
1539
1540         data
1541     end
1542   end
1543
1544   defp collection_private(%{"first" => %{"type" => type}})
1545        when type in ["CollectionPage", "OrderedCollectionPage"],
1546        do: {:ok, false}
1547
1548   defp collection_private(%{"first" => first}) do
1549     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1550            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1551       {:ok, false}
1552     else
1553       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1554       {:error, _} = e -> e
1555       e -> {:error, e}
1556     end
1557   end
1558
1559   defp collection_private(_data), do: {:ok, true}
1560
1561   def user_data_from_user_object(data) do
1562     with {:ok, data} <- MRF.filter(data),
1563          {:ok, data} <- object_to_user_data(data) do
1564       {:ok, data}
1565     else
1566       e -> {:error, e}
1567     end
1568   end
1569
1570   def fetch_and_prepare_user_from_ap_id(ap_id) do
1571     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1572          {:ok, data} <- user_data_from_user_object(data),
1573          data <- maybe_update_follow_information(data) do
1574       {:ok, data}
1575     else
1576       {:error, "Object has been deleted"} = e ->
1577         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1578         {:error, e}
1579
1580       e ->
1581         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1582         {:error, e}
1583     end
1584   end
1585
1586   def make_user_from_ap_id(ap_id) do
1587     user = User.get_cached_by_ap_id(ap_id)
1588
1589     if user && !User.ap_enabled?(user) do
1590       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1591     else
1592       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1593         if user do
1594           user
1595           |> User.remote_user_changeset(data)
1596           |> User.update_and_set_cache()
1597         else
1598           data
1599           |> User.remote_user_changeset()
1600           |> Repo.insert()
1601           |> User.set_cache()
1602         end
1603       else
1604         e -> {:error, e}
1605       end
1606     end
1607   end
1608
1609   def make_user_from_nickname(nickname) do
1610     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1611       make_user_from_ap_id(ap_id)
1612     else
1613       _e -> {:error, "No AP id in WebFinger"}
1614     end
1615   end
1616
1617   # filter out broken threads
1618   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1619     entire_thread_visible_for_user?(activity, user)
1620   end
1621
1622   # do post-processing on a specific activity
1623   def contain_activity(%Activity{} = activity, %User{} = user) do
1624     contain_broken_threads(activity, user)
1625   end
1626
1627   def fetch_direct_messages_query do
1628     Activity
1629     |> restrict_type(%{"type" => "Create"})
1630     |> restrict_visibility(%{visibility: "direct"})
1631     |> order_by([activity], asc: activity.id)
1632   end
1633 end