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     following_ap_ids =
 920       user
 921       |> User.get_friends_ap_ids()
 922
 923     query =
 924       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 925
 926     from(
 927       [activity, object: o] in query,
 928       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 929       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 930       where:
 931         fragment(
 932           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 933           activity.data,
 934           activity.data,
 935           ^blocks
 936         ),
 937       where:
 938         fragment(
 939           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
 940           activity.actor,
 941           ^domain_blocks,
 942           activity.actor,
 943           ^following_ap_ids
 944         ),
 945       where:
 946         fragment(
 947           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
 948           o.data,
 949           ^domain_blocks,
 950           o.data,
 951           ^following_ap_ids
 952         )
 953     )
 954   end
 955
 956   defp restrict_blocked(query, _), do: query
 957
 958   defp restrict_unlisted(query) do
 959     from(
 960       activity in query,
 961       where:
 962         fragment(
 963           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 964           activity.data,
 965           ^[Pleroma.Constants.as_public()]
 966         )
 967     )
 968   end
 969
 970   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 971     from(activity in query, where: activity.id in ^ids)
 972   end
 973
 974   defp restrict_pinned(query, _), do: query
 975
 976   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user}) do
 977     muted_reblogs = user.muted_reblogs || []
 978
 979     from(
 980       activity in query,
 981       where:
 982         fragment(
 983           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 984           activity.data,
 985           activity.actor,
 986           ^muted_reblogs
 987         )
 988     )
 989   end
 990
 991   defp restrict_muted_reblogs(query, _), do: query
 992
 993   defp restrict_instance(query, %{"instance" => instance}) do
 994     users =
 995       from(
 996         u in User,
 997         select: u.ap_id,
 998         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
 999       )
1000       |> Repo.all()
1001
1002     from(activity in query, where: activity.actor in ^users)
1003   end
1004
1005   defp restrict_instance(query, _), do: query
1006
1007   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1008
1009   defp exclude_poll_votes(query, _) do
1010     if has_named_binding?(query, :object) do
1011       from([activity, object: o] in query,
1012         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1013       )
1014     else
1015       query
1016     end
1017   end
1018
1019   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1020     from(activity in query, where: activity.id != ^id)
1021   end
1022
1023   defp exclude_id(query, _), do: query
1024
1025   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1026
1027   defp maybe_preload_objects(query, _) do
1028     query
1029     |> Activity.with_preloaded_object()
1030   end
1031
1032   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1033
1034   defp maybe_preload_bookmarks(query, opts) do
1035     query
1036     |> Activity.with_preloaded_bookmark(opts["user"])
1037   end
1038
1039   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1040
1041   defp maybe_set_thread_muted_field(query, opts) do
1042     query
1043     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1044   end
1045
1046   defp maybe_order(query, %{order: :desc}) do
1047     query
1048     |> order_by(desc: :id)
1049   end
1050
1051   defp maybe_order(query, %{order: :asc}) do
1052     query
1053     |> order_by(asc: :id)
1054   end
1055
1056   defp maybe_order(query, _), do: query
1057
1058   def fetch_activities_query(recipients, opts \\ %{}) do
1059     config = %{
1060       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1061     }
1062
1063     Activity
1064     |> maybe_preload_objects(opts)
1065     |> maybe_preload_bookmarks(opts)
1066     |> maybe_set_thread_muted_field(opts)
1067     |> maybe_order(opts)
1068     |> restrict_recipients(recipients, opts["user"])
1069     |> restrict_tag(opts)
1070     |> restrict_tag_reject(opts)
1071     |> restrict_tag_all(opts)
1072     |> restrict_since(opts)
1073     |> restrict_local(opts)
1074     |> restrict_actor(opts)
1075     |> restrict_type(opts)
1076     |> restrict_state(opts)
1077     |> restrict_favorited_by(opts)
1078     |> restrict_blocked(opts)
1079     |> restrict_muted(opts)
1080     |> restrict_media(opts)
1081     |> restrict_visibility(opts)
1082     |> restrict_thread_visibility(opts, config)
1083     |> restrict_replies(opts)
1084     |> restrict_reblogs(opts)
1085     |> restrict_pinned(opts)
1086     |> restrict_muted_reblogs(opts)
1087     |> restrict_instance(opts)
1088     |> Activity.restrict_deactivated_users()
1089     |> exclude_poll_votes(opts)
1090     |> exclude_visibility(opts)
1091   end
1092
1093   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1094     list_memberships = Pleroma.List.memberships(opts["user"])
1095
1096     fetch_activities_query(recipients ++ list_memberships, opts)
1097     |> Pagination.fetch_paginated(opts, pagination)
1098     |> Enum.reverse()
1099     |> maybe_update_cc(list_memberships, opts["user"])
1100   end
1101
1102   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1103        when is_list(list_memberships) and length(list_memberships) > 0 do
1104     Enum.map(activities, fn
1105       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1106         if Enum.any?(bcc, &(&1 in list_memberships)) do
1107           update_in(activity.data["cc"], &[user_ap_id | &1])
1108         else
1109           activity
1110         end
1111
1112       activity ->
1113         activity
1114     end)
1115   end
1116
1117   defp maybe_update_cc(activities, _, _), do: activities
1118
1119   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1120     from(activity in query,
1121       where:
1122         fragment("? && ?", activity.recipients, ^recipients) or
1123           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1124              ^Pleroma.Constants.as_public() in activity.recipients)
1125     )
1126   end
1127
1128   def fetch_activities_bounded(
1129         recipients,
1130         recipients_with_public,
1131         opts \\ %{},
1132         pagination \\ :keyset
1133       ) do
1134     fetch_activities_query([], opts)
1135     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1136     |> Pagination.fetch_paginated(opts, pagination)
1137     |> Enum.reverse()
1138   end
1139
1140   def upload(file, opts \\ []) do
1141     with {:ok, data} <- Upload.store(file, opts) do
1142       obj_data =
1143         if opts[:actor] do
1144           Map.put(data, "actor", opts[:actor])
1145         else
1146           data
1147         end
1148
1149       Repo.insert(%Object{data: obj_data})
1150     end
1151   end
1152
1153   defp object_to_user_data(data) do
1154     avatar =
1155       data["icon"]["url"] &&
1156         %{
1157           "type" => "Image",
1158           "url" => [%{"href" => data["icon"]["url"]}]
1159         }
1160
1161     banner =
1162       data["image"]["url"] &&
1163         %{
1164           "type" => "Image",
1165           "url" => [%{"href" => data["image"]["url"]}]
1166         }
1167
1168     fields =
1169       data
1170       |> Map.get("attachment", [])
1171       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1172       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1173
1174     locked = data["manuallyApprovesFollowers"] || false
1175     data = Transmogrifier.maybe_fix_user_object(data)
1176     discoverable = data["discoverable"] || false
1177     invisible = data["invisible"] || false
1178
1179     user_data = %{
1180       ap_id: data["id"],
1181       ap_enabled: true,
1182       source_data: data,
1183       banner: banner,
1184       fields: fields,
1185       locked: locked,
1186       discoverable: discoverable,
1187       invisible: invisible,
1188       avatar: avatar,
1189       name: data["name"],
1190       follower_address: data["followers"],
1191       following_address: data["following"],
1192       bio: data["summary"]
1193     }
1194
1195     # nickname can be nil because of virtual actors
1196     user_data =
1197       if data["preferredUsername"] do
1198         Map.put(
1199           user_data,
1200           :nickname,
1201           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1202         )
1203       else
1204         Map.put(user_data, :nickname, nil)
1205       end
1206
1207     {:ok, user_data}
1208   end
1209
1210   def fetch_follow_information_for_user(user) do
1211     with {:ok, following_data} <-
1212            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1213          following_count when is_integer(following_count) <- following_data["totalItems"],
1214          {:ok, hide_follows} <- collection_private(following_data),
1215          {:ok, followers_data} <-
1216            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1217          followers_count when is_integer(followers_count) <- followers_data["totalItems"],
1218          {:ok, hide_followers} <- collection_private(followers_data) do
1219       {:ok,
1220        %{
1221          hide_follows: hide_follows,
1222          follower_count: followers_count,
1223          following_count: following_count,
1224          hide_followers: hide_followers
1225        }}
1226     else
1227       {:error, _} = e ->
1228         e
1229
1230       e ->
1231         {:error, e}
1232     end
1233   end
1234
1235   defp maybe_update_follow_information(data) do
1236     with {:enabled, true} <-
1237            {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
1238          {:ok, info} <- fetch_follow_information_for_user(data) do
1239       info = Map.merge(data[:info] || %{}, info)
1240       Map.put(data, :info, info)
1241     else
1242       {:enabled, false} ->
1243         data
1244
1245       e ->
1246         Logger.error(
1247           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1248         )
1249
1250         data
1251     end
1252   end
1253
1254   defp collection_private(data) do
1255     if is_map(data["first"]) and
1256          data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
1257       {:ok, false}
1258     else
1259       with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1260              Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
1261         {:ok, false}
1262       else
1263         {:error, {:ok, %{status: code}}} when code in [401, 403] ->
1264           {:ok, true}
1265
1266         {:error, _} = e ->
1267           e
1268
1269         e ->
1270           {:error, e}
1271       end
1272     end
1273   end
1274
1275   def user_data_from_user_object(data) do
1276     with {:ok, data} <- MRF.filter(data),
1277          {:ok, data} <- object_to_user_data(data) do
1278       {:ok, data}
1279     else
1280       e -> {:error, e}
1281     end
1282   end
1283
1284   def fetch_and_prepare_user_from_ap_id(ap_id) do
1285     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1286          {:ok, data} <- user_data_from_user_object(data),
1287          data <- maybe_update_follow_information(data) do
1288       {:ok, data}
1289     else
1290       e ->
1291         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1292         {:error, e}
1293     end
1294   end
1295
1296   def make_user_from_ap_id(ap_id) do
1297     if _user = User.get_cached_by_ap_id(ap_id) do
1298       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1299     else
1300       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1301         User.insert_or_update_user(data)
1302       else
1303         e -> {:error, e}
1304       end
1305     end
1306   end
1307
1308   def make_user_from_nickname(nickname) do
1309     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1310       make_user_from_ap_id(ap_id)
1311     else
1312       _e -> {:error, "No AP id in WebFinger"}
1313     end
1314   end
1315
1316   # filter out broken threads
1317   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1318     entire_thread_visible_for_user?(activity, user)
1319   end
1320
1321   # do post-processing on a specific activity
1322   def contain_activity(%Activity{} = activity, %User{} = user) do
1323     contain_broken_threads(activity, user)
1324   end
1325
1326   def fetch_direct_messages_query do
1327     Activity
1328     |> restrict_type(%{"type" => "Create"})
1329     |> restrict_visibility(%{visibility: "direct"})
1330     |> order_by([activity], asc: activity.id)
1331   end
1332 end