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   defp user_activities_recipients(%{"godmode" => true}) do
 738     []
 739   end
 740
 741   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 742     if reading_user do
 743       [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 744     else
 745       [Pleroma.Constants.as_public()]
 746     end
 747   end
 748
 749   defp restrict_since(query, %{"since_id" => ""}), do: query
 750
 751   defp restrict_since(query, %{"since_id" => since_id}) do
 752     from(activity in query, where: activity.id > ^since_id)
 753   end
 754
 755   defp restrict_since(query, _), do: query
 756
 757   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 758     raise "Can't use the child object without preloading!"
 759   end
 760
 761   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 762        when is_list(tag_reject) and tag_reject != [] do
 763     from(
 764       [_activity, object] in query,
 765       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 766     )
 767   end
 768
 769   defp restrict_tag_reject(query, _), do: query
 770
 771   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 772     raise "Can't use the child object without preloading!"
 773   end
 774
 775   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 776        when is_list(tag_all) and tag_all != [] do
 777     from(
 778       [_activity, object] in query,
 779       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 780     )
 781   end
 782
 783   defp restrict_tag_all(query, _), do: query
 784
 785   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 786     raise "Can't use the child object without preloading!"
 787   end
 788
 789   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 790     from(
 791       [_activity, object] in query,
 792       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 793     )
 794   end
 795
 796   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 797     from(
 798       [_activity, object] in query,
 799       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 800     )
 801   end
 802
 803   defp restrict_tag(query, _), do: query
 804
 805   defp restrict_recipients(query, [], _user), do: query
 806
 807   defp restrict_recipients(query, recipients, nil) do
 808     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 809   end
 810
 811   defp restrict_recipients(query, recipients, user) do
 812     from(
 813       activity in query,
 814       where: fragment("? && ?", ^recipients, activity.recipients),
 815       or_where: activity.actor == ^user.ap_id
 816     )
 817   end
 818
 819   defp restrict_local(query, %{"local_only" => true}) do
 820     from(activity in query, where: activity.local == true)
 821   end
 822
 823   defp restrict_local(query, _), do: query
 824
 825   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 826     from(activity in query, where: activity.actor == ^actor_id)
 827   end
 828
 829   defp restrict_actor(query, _), do: query
 830
 831   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 832     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 833   end
 834
 835   defp restrict_type(query, %{"type" => type}) do
 836     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 837   end
 838
 839   defp restrict_type(query, _), do: query
 840
 841   defp restrict_state(query, %{"state" => state}) do
 842     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 843   end
 844
 845   defp restrict_state(query, _), do: query
 846
 847   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 848     from(
 849       [_activity, object] in query,
 850       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 851     )
 852   end
 853
 854   defp restrict_favorited_by(query, _), do: query
 855
 856   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 857     raise "Can't use the child object without preloading!"
 858   end
 859
 860   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 861     from(
 862       [_activity, object] in query,
 863       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 864     )
 865   end
 866
 867   defp restrict_media(query, _), do: query
 868
 869   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 870     from(
 871       [_activity, object] in query,
 872       where: fragment("?->>'inReplyTo' is null", object.data)
 873     )
 874   end
 875
 876   defp restrict_replies(query, _), do: query
 877
 878   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 879     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 880   end
 881
 882   defp restrict_reblogs(query, _), do: query
 883
 884   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 885
 886   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
 887     mutes = opts["muted_ap_ids"] || User.muted_ap_ids(user)
 888
 889     query =
 890       from([activity] in query,
 891         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 892         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 893       )
 894
 895     unless opts["skip_preload"] do
 896       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
 897     else
 898       query
 899     end
 900   end
 901
 902   defp restrict_muted(query, _), do: query
 903
 904   defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
 905     blocked_ap_ids = opts["blocked_ap_ids"] || User.blocked_ap_ids(user)
 906     domain_blocks = user.domain_blocks || []
 907
 908     query =
 909       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 910
 911     from(
 912       [activity, object: o] in query,
 913       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
 914       where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
 915       where:
 916         fragment(
 917           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 918           activity.data,
 919           activity.data,
 920           ^blocked_ap_ids
 921         ),
 922       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 923       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 924     )
 925   end
 926
 927   defp restrict_blocked(query, _), do: query
 928
 929   defp restrict_unlisted(query) do
 930     from(
 931       activity in query,
 932       where:
 933         fragment(
 934           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 935           activity.data,
 936           ^[Pleroma.Constants.as_public()]
 937         )
 938     )
 939   end
 940
 941   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 942     from(activity in query, where: activity.id in ^ids)
 943   end
 944
 945   defp restrict_pinned(query, _), do: query
 946
 947   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user}) do
 948     muted_reblogs = user.muted_reblogs || []
 949
 950     from(
 951       activity in query,
 952       where:
 953         fragment(
 954           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 955           activity.data,
 956           activity.actor,
 957           ^muted_reblogs
 958         )
 959     )
 960   end
 961
 962   defp restrict_muted_reblogs(query, _), do: query
 963
 964   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
 965
 966   defp exclude_poll_votes(query, _) do
 967     if has_named_binding?(query, :object) do
 968       from([activity, object: o] in query,
 969         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 970       )
 971     else
 972       query
 973     end
 974   end
 975
 976   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
 977     from(activity in query, where: activity.id != ^id)
 978   end
 979
 980   defp exclude_id(query, _), do: query
 981
 982   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 983
 984   defp maybe_preload_objects(query, _) do
 985     query
 986     |> Activity.with_preloaded_object()
 987   end
 988
 989   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 990
 991   defp maybe_preload_bookmarks(query, opts) do
 992     query
 993     |> Activity.with_preloaded_bookmark(opts["user"])
 994   end
 995
 996   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 997
 998   defp maybe_set_thread_muted_field(query, opts) do
 999     query
1000     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1001   end
1002
1003   defp maybe_order(query, %{order: :desc}) do
1004     query
1005     |> order_by(desc: :id)
1006   end
1007
1008   defp maybe_order(query, %{order: :asc}) do
1009     query
1010     |> order_by(asc: :id)
1011   end
1012
1013   defp maybe_order(query, _), do: query
1014
1015   def fetch_activities_query(recipients, opts \\ %{}) do
1016     config = %{
1017       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1018     }
1019
1020     Activity
1021     |> maybe_preload_objects(opts)
1022     |> maybe_preload_bookmarks(opts)
1023     |> maybe_set_thread_muted_field(opts)
1024     |> maybe_order(opts)
1025     |> restrict_recipients(recipients, opts["user"])
1026     |> restrict_tag(opts)
1027     |> restrict_tag_reject(opts)
1028     |> restrict_tag_all(opts)
1029     |> restrict_since(opts)
1030     |> restrict_local(opts)
1031     |> restrict_actor(opts)
1032     |> restrict_type(opts)
1033     |> restrict_state(opts)
1034     |> restrict_favorited_by(opts)
1035     |> restrict_blocked(opts)
1036     |> restrict_muted(opts)
1037     |> restrict_media(opts)
1038     |> restrict_visibility(opts)
1039     |> restrict_thread_visibility(opts, config)
1040     |> restrict_replies(opts)
1041     |> restrict_reblogs(opts)
1042     |> restrict_pinned(opts)
1043     |> restrict_muted_reblogs(opts)
1044     |> Activity.restrict_deactivated_users()
1045     |> exclude_poll_votes(opts)
1046     |> exclude_visibility(opts)
1047   end
1048
1049   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1050     list_memberships = Pleroma.List.memberships(opts["user"])
1051
1052     fetch_activities_query(recipients ++ list_memberships, opts)
1053     |> Pagination.fetch_paginated(opts, pagination)
1054     |> Enum.reverse()
1055     |> maybe_update_cc(list_memberships, opts["user"])
1056   end
1057
1058   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1059        when is_list(list_memberships) and length(list_memberships) > 0 do
1060     Enum.map(activities, fn
1061       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1062         if Enum.any?(bcc, &(&1 in list_memberships)) do
1063           update_in(activity.data["cc"], &[user_ap_id | &1])
1064         else
1065           activity
1066         end
1067
1068       activity ->
1069         activity
1070     end)
1071   end
1072
1073   defp maybe_update_cc(activities, _, _), do: activities
1074
1075   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1076     from(activity in query,
1077       where:
1078         fragment("? && ?", activity.recipients, ^recipients) or
1079           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1080              ^Pleroma.Constants.as_public() in activity.recipients)
1081     )
1082   end
1083
1084   def fetch_activities_bounded(
1085         recipients,
1086         recipients_with_public,
1087         opts \\ %{},
1088         pagination \\ :keyset
1089       ) do
1090     fetch_activities_query([], opts)
1091     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1092     |> Pagination.fetch_paginated(opts, pagination)
1093     |> Enum.reverse()
1094   end
1095
1096   def upload(file, opts \\ []) do
1097     with {:ok, data} <- Upload.store(file, opts) do
1098       obj_data =
1099         if opts[:actor] do
1100           Map.put(data, "actor", opts[:actor])
1101         else
1102           data
1103         end
1104
1105       Repo.insert(%Object{data: obj_data})
1106     end
1107   end
1108
1109   defp object_to_user_data(data) do
1110     avatar =
1111       data["icon"]["url"] &&
1112         %{
1113           "type" => "Image",
1114           "url" => [%{"href" => data["icon"]["url"]}]
1115         }
1116
1117     banner =
1118       data["image"]["url"] &&
1119         %{
1120           "type" => "Image",
1121           "url" => [%{"href" => data["image"]["url"]}]
1122         }
1123
1124     fields =
1125       data
1126       |> Map.get("attachment", [])
1127       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1128       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1129
1130     locked = data["manuallyApprovesFollowers"] || false
1131     data = Transmogrifier.maybe_fix_user_object(data)
1132     discoverable = data["discoverable"] || false
1133     invisible = data["invisible"] || false
1134
1135     user_data = %{
1136       ap_id: data["id"],
1137       ap_enabled: true,
1138       source_data: data,
1139       banner: banner,
1140       fields: fields,
1141       locked: locked,
1142       discoverable: discoverable,
1143       invisible: invisible,
1144       avatar: avatar,
1145       name: data["name"],
1146       follower_address: data["followers"],
1147       following_address: data["following"],
1148       bio: data["summary"]
1149     }
1150
1151     # nickname can be nil because of virtual actors
1152     user_data =
1153       if data["preferredUsername"] do
1154         Map.put(
1155           user_data,
1156           :nickname,
1157           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1158         )
1159       else
1160         Map.put(user_data, :nickname, nil)
1161       end
1162
1163     {:ok, user_data}
1164   end
1165
1166   def fetch_follow_information_for_user(user) do
1167     with {:ok, following_data} <-
1168            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1169          following_count when is_integer(following_count) <- following_data["totalItems"],
1170          {:ok, hide_follows} <- collection_private(following_data),
1171          {:ok, followers_data} <-
1172            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1173          followers_count when is_integer(followers_count) <- followers_data["totalItems"],
1174          {:ok, hide_followers} <- collection_private(followers_data) do
1175       {:ok,
1176        %{
1177          hide_follows: hide_follows,
1178          follower_count: followers_count,
1179          following_count: following_count,
1180          hide_followers: hide_followers
1181        }}
1182     else
1183       {:error, _} = e ->
1184         e
1185
1186       e ->
1187         {:error, e}
1188     end
1189   end
1190
1191   defp maybe_update_follow_information(data) do
1192     with {:enabled, true} <-
1193            {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
1194          {:ok, info} <- fetch_follow_information_for_user(data) do
1195       info = Map.merge(data[:info] || %{}, info)
1196       Map.put(data, :info, info)
1197     else
1198       {:enabled, false} ->
1199         data
1200
1201       e ->
1202         Logger.error(
1203           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1204         )
1205
1206         data
1207     end
1208   end
1209
1210   defp collection_private(data) do
1211     if is_map(data["first"]) and
1212          data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
1213       {:ok, false}
1214     else
1215       with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1216              Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
1217         {:ok, false}
1218       else
1219         {:error, {:ok, %{status: code}}} when code in [401, 403] ->
1220           {:ok, true}
1221
1222         {:error, _} = e ->
1223           e
1224
1225         e ->
1226           {:error, e}
1227       end
1228     end
1229   end
1230
1231   def user_data_from_user_object(data) do
1232     with {:ok, data} <- MRF.filter(data),
1233          {:ok, data} <- object_to_user_data(data) do
1234       {:ok, data}
1235     else
1236       e -> {:error, e}
1237     end
1238   end
1239
1240   def fetch_and_prepare_user_from_ap_id(ap_id) do
1241     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1242          {:ok, data} <- user_data_from_user_object(data),
1243          data <- maybe_update_follow_information(data) do
1244       {:ok, data}
1245     else
1246       e ->
1247         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1248         {:error, e}
1249     end
1250   end
1251
1252   def make_user_from_ap_id(ap_id) do
1253     if _user = User.get_cached_by_ap_id(ap_id) do
1254       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1255     else
1256       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1257         User.insert_or_update_user(data)
1258       else
1259         e -> {:error, e}
1260       end
1261     end
1262   end
1263
1264   def make_user_from_nickname(nickname) do
1265     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1266       make_user_from_ap_id(ap_id)
1267     else
1268       _e -> {:error, "No AP id in WebFinger"}
1269     end
1270   end
1271
1272   # filter out broken threads
1273   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1274     entire_thread_visible_for_user?(activity, user)
1275   end
1276
1277   # do post-processing on a specific activity
1278   def contain_activity(%Activity{} = activity, %User{} = user) do
1279     contain_broken_threads(activity, user)
1280   end
1281
1282   def fetch_direct_messages_query do
1283     Activity
1284     |> restrict_type(%{"type" => "Create"})
1285     |> restrict_visibility(%{visibility: "direct"})
1286     |> order_by([activity], asc: activity.id)
1287   end
1288 end