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

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