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   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 326   def like(
 327         %User{ap_id: ap_id} = user,
 328         %Object{data: %{"id" => _}} = object,
 329         activity_id \\ nil,
 330         local \\ true
 331       ) do
 332     with nil <- get_existing_like(ap_id, object),
 333          like_data <- make_like_data(user, object, activity_id),
 334          {:ok, activity} <- insert(like_data, local),
 335          {:ok, object} <- add_like_to_object(activity, object),
 336          :ok <- maybe_federate(activity) do
 337       {:ok, activity, object}
 338     else
 339       %Activity{} = activity -> {:ok, activity, object}
 340       error -> {:error, error}
 341     end
 342   end
 343
 344   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
 345     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 346          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 347          {:ok, unlike_activity} <- insert(unlike_data, local),
 348          {:ok, _activity} <- Repo.delete(like_activity),
 349          {:ok, object} <- remove_like_from_object(like_activity, object),
 350          :ok <- maybe_federate(unlike_activity) do
 351       {:ok, unlike_activity, like_activity, object}
 352     else
 353       _e -> {:ok, object}
 354     end
 355   end
 356
 357   def announce(
 358         %User{ap_id: _} = user,
 359         %Object{data: %{"id" => _}} = object,
 360         activity_id \\ nil,
 361         local \\ true,
 362         public \\ true
 363       ) do
 364     with true <- is_announceable?(object, user, public),
 365          announce_data <- make_announce_data(user, object, activity_id, public),
 366          {:ok, activity} <- insert(announce_data, local),
 367          {:ok, object} <- add_announce_to_object(activity, object),
 368          :ok <- maybe_federate(activity) do
 369       {:ok, activity, object}
 370     else
 371       error -> {:error, error}
 372     end
 373   end
 374
 375   def unannounce(
 376         %User{} = actor,
 377         %Object{} = object,
 378         activity_id \\ nil,
 379         local \\ true
 380       ) do
 381     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 382          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 383          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 384          :ok <- maybe_federate(unannounce_activity),
 385          {:ok, _activity} <- Repo.delete(announce_activity),
 386          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 387       {:ok, unannounce_activity, object}
 388     else
 389       _e -> {:ok, object}
 390     end
 391   end
 392
 393   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 394     with data <- make_follow_data(follower, followed, activity_id),
 395          {:ok, activity} <- insert(data, local),
 396          :ok <- maybe_federate(activity),
 397          _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
 398       {:ok, activity}
 399     end
 400   end
 401
 402   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 403     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 404          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 405          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 406          {:ok, activity} <- insert(unfollow_data, local),
 407          :ok <- maybe_federate(activity) do
 408       {:ok, activity}
 409     end
 410   end
 411
 412   def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
 413     with data <- %{
 414            "to" => [follower_address],
 415            "type" => "Delete",
 416            "actor" => ap_id,
 417            "object" => %{"type" => "Person", "id" => ap_id}
 418          },
 419          {:ok, activity} <- insert(data, true, true, true),
 420          :ok <- maybe_federate(activity) do
 421       {:ok, user}
 422     end
 423   end
 424
 425   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options \\ []) do
 426     local = Keyword.get(options, :local, true)
 427     activity_id = Keyword.get(options, :activity_id, nil)
 428     actor = Keyword.get(options, :actor, actor)
 429
 430     user = User.get_cached_by_ap_id(actor)
 431     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 432
 433     with {:ok, object, activity} <- Object.delete(object),
 434          data <-
 435            %{
 436              "type" => "Delete",
 437              "actor" => actor,
 438              "object" => id,
 439              "to" => to,
 440              "deleted_activity_id" => activity && activity.id
 441            }
 442            |> maybe_put("id", activity_id),
 443          {:ok, activity} <- insert(data, local, false),
 444          stream_out_participations(object, user),
 445          _ <- decrease_replies_count_if_reply(object),
 446          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 447          :ok <- maybe_federate(activity) do
 448       {:ok, activity}
 449     end
 450   end
 451
 452   @spec block(User.t(), User.t(), String.t() | nil, boolean) :: {:ok, Activity.t() | nil}
 453   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 454     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
 455     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 456
 457     if unfollow_blocked do
 458       follow_activity = fetch_latest_follow(blocker, blocked)
 459       if follow_activity, do: unfollow(blocker, blocked, nil, local)
 460     end
 461
 462     with true <- outgoing_blocks,
 463          block_data <- make_block_data(blocker, blocked, activity_id),
 464          {:ok, activity} <- insert(block_data, local),
 465          :ok <- maybe_federate(activity) do
 466       {:ok, activity}
 467     else
 468       _e -> {:ok, nil}
 469     end
 470   end
 471
 472   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 473     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 474          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 475          {:ok, activity} <- insert(unblock_data, local),
 476          :ok <- maybe_federate(activity) do
 477       {:ok, activity}
 478     end
 479   end
 480
 481   @spec flag(map()) :: {:ok, Activity.t()} | any
 482   def flag(
 483         %{
 484           actor: actor,
 485           context: _context,
 486           account: account,
 487           statuses: statuses,
 488           content: content
 489         } = params
 490       ) do
 491     # only accept false as false value
 492     local = !(params[:local] == false)
 493     forward = !(params[:forward] == false)
 494
 495     additional = params[:additional] || %{}
 496
 497     additional =
 498       if forward do
 499         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 500       else
 501         Map.merge(additional, %{"to" => [], "cc" => []})
 502       end
 503
 504     with flag_data <- make_flag_data(params, additional),
 505          {:ok, activity} <- insert(flag_data, local),
 506          {:ok, stripped_activity} <- strip_report_status_data(activity),
 507          :ok <- maybe_federate(stripped_activity) do
 508       Enum.each(User.all_superusers(), fn superuser ->
 509         superuser
 510         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 511         |> Pleroma.Emails.Mailer.deliver_async()
 512       end)
 513
 514       {:ok, activity}
 515     end
 516   end
 517
 518   defp fetch_activities_for_context_query(context, opts) do
 519     public = [Pleroma.Constants.as_public()]
 520
 521     recipients =
 522       if opts["user"],
 523         do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
 524         else: public
 525
 526     from(activity in Activity)
 527     |> maybe_preload_objects(opts)
 528     |> maybe_preload_bookmarks(opts)
 529     |> maybe_set_thread_muted_field(opts)
 530     |> restrict_blocked(opts)
 531     |> restrict_recipients(recipients, opts["user"])
 532     |> where(
 533       [activity],
 534       fragment(
 535         "?->>'type' = ? and ?->>'context' = ?",
 536         activity.data,
 537         "Create",
 538         activity.data,
 539         ^context
 540       )
 541     )
 542     |> exclude_poll_votes(opts)
 543     |> exclude_id(opts)
 544     |> order_by([activity], desc: activity.id)
 545   end
 546
 547   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 548   def fetch_activities_for_context(context, opts \\ %{}) do
 549     context
 550     |> fetch_activities_for_context_query(opts)
 551     |> Repo.all()
 552   end
 553
 554   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 555           FlakeId.Ecto.CompatType.t() | nil
 556   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 557     context
 558     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 559     |> limit(1)
 560     |> select([a], a.id)
 561     |> Repo.one()
 562   end
 563
 564   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 565     opts = Map.drop(opts, ["user"])
 566
 567     [Pleroma.Constants.as_public()]
 568     |> fetch_activities_query(opts)
 569     |> restrict_unlisted()
 570     |> Pagination.fetch_paginated(opts, pagination)
 571     |> Enum.reverse()
 572   end
 573
 574   @valid_visibilities ~w[direct unlisted public private]
 575
 576   defp restrict_visibility(query, %{visibility: visibility})
 577        when is_list(visibility) do
 578     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 579       query =
 580         from(
 581           a in query,
 582           where:
 583             fragment(
 584               "activity_visibility(?, ?, ?) = ANY (?)",
 585               a.actor,
 586               a.recipients,
 587               a.data,
 588               ^visibility
 589             )
 590         )
 591
 592       query
 593     else
 594       Logger.error("Could not restrict visibility to #{visibility}")
 595     end
 596   end
 597
 598   defp restrict_visibility(query, %{visibility: visibility})
 599        when visibility in @valid_visibilities do
 600     from(
 601       a in query,
 602       where:
 603         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 604     )
 605   end
 606
 607   defp restrict_visibility(_query, %{visibility: visibility})
 608        when visibility not in @valid_visibilities do
 609     Logger.error("Could not restrict visibility to #{visibility}")
 610   end
 611
 612   defp restrict_visibility(query, _visibility), do: query
 613
 614   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 615        when is_list(visibility) do
 616     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 617       from(
 618         a in query,
 619         where:
 620           not fragment(
 621             "activity_visibility(?, ?, ?) = ANY (?)",
 622             a.actor,
 623             a.recipients,
 624             a.data,
 625             ^visibility
 626           )
 627       )
 628     else
 629       Logger.error("Could not exclude visibility to #{visibility}")
 630       query
 631     end
 632   end
 633
 634   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 635        when visibility in @valid_visibilities do
 636     from(
 637       a in query,
 638       where:
 639         not fragment(
 640           "activity_visibility(?, ?, ?) = ?",
 641           a.actor,
 642           a.recipients,
 643           a.data,
 644           ^visibility
 645         )
 646     )
 647   end
 648
 649   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 650        when visibility not in @valid_visibilities do
 651     Logger.error("Could not exclude visibility to #{visibility}")
 652     query
 653   end
 654
 655   defp exclude_visibility(query, _visibility), do: query
 656
 657   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 658     do: query
 659
 660   defp restrict_thread_visibility(
 661          query,
 662          %{"user" => %User{skip_thread_containment: true}},
 663          _
 664        ),
 665        do: query
 666
 667   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 668     from(
 669       a in query,
 670       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 671     )
 672   end
 673
 674   defp restrict_thread_visibility(query, _, _), do: query
 675
 676   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 677     params =
 678       params
 679       |> Map.put("user", reading_user)
 680       |> Map.put("actor_id", user.ap_id)
 681       |> Map.put("whole_db", true)
 682
 683     recipients =
 684       user_activities_recipients(%{
 685         "godmode" => params["godmode"],
 686         "reading_user" => reading_user
 687       })
 688
 689     fetch_activities(recipients, params)
 690     |> Enum.reverse()
 691   end
 692
 693   def fetch_user_activities(user, reading_user, params \\ %{}) do
 694     params =
 695       params
 696       |> Map.put("type", ["Create", "Announce"])
 697       |> Map.put("user", reading_user)
 698       |> Map.put("actor_id", user.ap_id)
 699       |> Map.put("whole_db", true)
 700       |> Map.put("pinned_activity_ids", user.pinned_activities)
 701
 702     recipients =
 703       user_activities_recipients(%{
 704         "godmode" => params["godmode"],
 705         "reading_user" => reading_user
 706       })
 707
 708     fetch_activities(recipients, params)
 709     |> Enum.reverse()
 710   end
 711
 712   defp user_activities_recipients(%{"godmode" => true}) do
 713     []
 714   end
 715
 716   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 717     if reading_user do
 718       [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 719     else
 720       [Pleroma.Constants.as_public()]
 721     end
 722   end
 723
 724   defp restrict_since(query, %{"since_id" => ""}), do: query
 725
 726   defp restrict_since(query, %{"since_id" => since_id}) do
 727     from(activity in query, where: activity.id > ^since_id)
 728   end
 729
 730   defp restrict_since(query, _), do: query
 731
 732   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 733     raise "Can't use the child object without preloading!"
 734   end
 735
 736   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 737        when is_list(tag_reject) and tag_reject != [] do
 738     from(
 739       [_activity, object] in query,
 740       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 741     )
 742   end
 743
 744   defp restrict_tag_reject(query, _), do: query
 745
 746   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 747     raise "Can't use the child object without preloading!"
 748   end
 749
 750   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 751        when is_list(tag_all) and tag_all != [] do
 752     from(
 753       [_activity, object] in query,
 754       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 755     )
 756   end
 757
 758   defp restrict_tag_all(query, _), do: query
 759
 760   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 761     raise "Can't use the child object without preloading!"
 762   end
 763
 764   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 765     from(
 766       [_activity, object] in query,
 767       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 768     )
 769   end
 770
 771   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 772     from(
 773       [_activity, object] in query,
 774       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 775     )
 776   end
 777
 778   defp restrict_tag(query, _), do: query
 779
 780   defp restrict_recipients(query, [], _user), do: query
 781
 782   defp restrict_recipients(query, recipients, nil) do
 783     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 784   end
 785
 786   defp restrict_recipients(query, recipients, user) do
 787     from(
 788       activity in query,
 789       where: fragment("? && ?", ^recipients, activity.recipients),
 790       or_where: activity.actor == ^user.ap_id
 791     )
 792   end
 793
 794   defp restrict_local(query, %{"local_only" => true}) do
 795     from(activity in query, where: activity.local == true)
 796   end
 797
 798   defp restrict_local(query, _), do: query
 799
 800   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 801     from(activity in query, where: activity.actor == ^actor_id)
 802   end
 803
 804   defp restrict_actor(query, _), do: query
 805
 806   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 807     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 808   end
 809
 810   defp restrict_type(query, %{"type" => type}) do
 811     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 812   end
 813
 814   defp restrict_type(query, _), do: query
 815
 816   defp restrict_state(query, %{"state" => state}) do
 817     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 818   end
 819
 820   defp restrict_state(query, _), do: query
 821
 822   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 823     from(
 824       [_activity, object] in query,
 825       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 826     )
 827   end
 828
 829   defp restrict_favorited_by(query, _), do: query
 830
 831   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 832     raise "Can't use the child object without preloading!"
 833   end
 834
 835   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 836     from(
 837       [_activity, object] in query,
 838       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 839     )
 840   end
 841
 842   defp restrict_media(query, _), do: query
 843
 844   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 845     from(
 846       [_activity, object] in query,
 847       where: fragment("?->>'inReplyTo' is null", object.data)
 848     )
 849   end
 850
 851   defp restrict_replies(query, _), do: query
 852
 853   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 854     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 855   end
 856
 857   defp restrict_reblogs(query, _), do: query
 858
 859   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 860
 861   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
 862     mutes = user.mutes
 863
 864     query =
 865       from([activity] in query,
 866         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 867         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 868       )
 869
 870     unless opts["skip_preload"] do
 871       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
 872     else
 873       query
 874     end
 875   end
 876
 877   defp restrict_muted(query, _), do: query
 878
 879   defp restrict_blocked(query, %{"blocking_user" => %User{} = user}) do
 880     blocks = user.blocks || []
 881     domain_blocks = user.domain_blocks || []
 882
 883     query =
 884       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 885
 886     from(
 887       [activity, object: o] in query,
 888       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 889       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 890       where:
 891         fragment(
 892           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 893           activity.data,
 894           activity.data,
 895           ^blocks
 896         ),
 897       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 898       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 899     )
 900   end
 901
 902   defp restrict_blocked(query, _), do: query
 903
 904   defp restrict_unlisted(query) do
 905     from(
 906       activity in query,
 907       where:
 908         fragment(
 909           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 910           activity.data,
 911           ^[Pleroma.Constants.as_public()]
 912         )
 913     )
 914   end
 915
 916   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 917     from(activity in query, where: activity.id in ^ids)
 918   end
 919
 920   defp restrict_pinned(query, _), do: query
 921
 922   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user}) do
 923     muted_reblogs = user.muted_reblogs || []
 924
 925     from(
 926       activity in query,
 927       where:
 928         fragment(
 929           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 930           activity.data,
 931           activity.actor,
 932           ^muted_reblogs
 933         )
 934     )
 935   end
 936
 937   defp restrict_muted_reblogs(query, _), do: query
 938
 939   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
 940
 941   defp exclude_poll_votes(query, _) do
 942     if has_named_binding?(query, :object) do
 943       from([activity, object: o] in query,
 944         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 945       )
 946     else
 947       query
 948     end
 949   end
 950
 951   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
 952     from(activity in query, where: activity.id != ^id)
 953   end
 954
 955   defp exclude_id(query, _), do: query
 956
 957   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 958
 959   defp maybe_preload_objects(query, _) do
 960     query
 961     |> Activity.with_preloaded_object()
 962   end
 963
 964   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 965
 966   defp maybe_preload_bookmarks(query, opts) do
 967     query
 968     |> Activity.with_preloaded_bookmark(opts["user"])
 969   end
 970
 971   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 972
 973   defp maybe_set_thread_muted_field(query, opts) do
 974     query
 975     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
 976   end
 977
 978   defp maybe_order(query, %{order: :desc}) do
 979     query
 980     |> order_by(desc: :id)
 981   end
 982
 983   defp maybe_order(query, %{order: :asc}) do
 984     query
 985     |> order_by(asc: :id)
 986   end
 987
 988   defp maybe_order(query, _), do: query
 989
 990   def fetch_activities_query(recipients, opts \\ %{}) do
 991     config = %{
 992       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
 993     }
 994
 995     Activity
 996     |> maybe_preload_objects(opts)
 997     |> maybe_preload_bookmarks(opts)
 998     |> maybe_set_thread_muted_field(opts)
 999     |> maybe_order(opts)
1000     |> restrict_recipients(recipients, opts["user"])
1001     |> restrict_tag(opts)
1002     |> restrict_tag_reject(opts)
1003     |> restrict_tag_all(opts)
1004     |> restrict_since(opts)
1005     |> restrict_local(opts)
1006     |> restrict_actor(opts)
1007     |> restrict_type(opts)
1008     |> restrict_state(opts)
1009     |> restrict_favorited_by(opts)
1010     |> restrict_blocked(opts)
1011     |> restrict_muted(opts)
1012     |> restrict_media(opts)
1013     |> restrict_visibility(opts)
1014     |> restrict_thread_visibility(opts, config)
1015     |> restrict_replies(opts)
1016     |> restrict_reblogs(opts)
1017     |> restrict_pinned(opts)
1018     |> restrict_muted_reblogs(opts)
1019     |> Activity.restrict_deactivated_users()
1020     |> exclude_poll_votes(opts)
1021     |> exclude_visibility(opts)
1022   end
1023
1024   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1025     list_memberships = Pleroma.List.memberships(opts["user"])
1026
1027     fetch_activities_query(recipients ++ list_memberships, opts)
1028     |> Pagination.fetch_paginated(opts, pagination)
1029     |> Enum.reverse()
1030     |> maybe_update_cc(list_memberships, opts["user"])
1031   end
1032
1033   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1034        when is_list(list_memberships) and length(list_memberships) > 0 do
1035     Enum.map(activities, fn
1036       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1037         if Enum.any?(bcc, &(&1 in list_memberships)) do
1038           update_in(activity.data["cc"], &[user_ap_id | &1])
1039         else
1040           activity
1041         end
1042
1043       activity ->
1044         activity
1045     end)
1046   end
1047
1048   defp maybe_update_cc(activities, _, _), do: activities
1049
1050   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1051     from(activity in query,
1052       where:
1053         fragment("? && ?", activity.recipients, ^recipients) or
1054           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1055              ^Pleroma.Constants.as_public() in activity.recipients)
1056     )
1057   end
1058
1059   def fetch_activities_bounded(
1060         recipients,
1061         recipients_with_public,
1062         opts \\ %{},
1063         pagination \\ :keyset
1064       ) do
1065     fetch_activities_query([], opts)
1066     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1067     |> Pagination.fetch_paginated(opts, pagination)
1068     |> Enum.reverse()
1069   end
1070
1071   def upload(file, opts \\ []) do
1072     with {:ok, data} <- Upload.store(file, opts) do
1073       obj_data =
1074         if opts[:actor] do
1075           Map.put(data, "actor", opts[:actor])
1076         else
1077           data
1078         end
1079
1080       Repo.insert(%Object{data: obj_data})
1081     end
1082   end
1083
1084   defp object_to_user_data(data) do
1085     avatar =
1086       data["icon"]["url"] &&
1087         %{
1088           "type" => "Image",
1089           "url" => [%{"href" => data["icon"]["url"]}]
1090         }
1091
1092     banner =
1093       data["image"]["url"] &&
1094         %{
1095           "type" => "Image",
1096           "url" => [%{"href" => data["image"]["url"]}]
1097         }
1098
1099     fields =
1100       data
1101       |> Map.get("attachment", [])
1102       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1103       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1104
1105     locked = data["manuallyApprovesFollowers"] || false
1106     data = Transmogrifier.maybe_fix_user_object(data)
1107     discoverable = data["discoverable"] || false
1108     invisible = data["invisible"] || false
1109
1110     user_data = %{
1111       ap_id: data["id"],
1112       ap_enabled: true,
1113       source_data: data,
1114       banner: banner,
1115       fields: fields,
1116       locked: locked,
1117       discoverable: discoverable,
1118       invisible: invisible,
1119       avatar: avatar,
1120       name: data["name"],
1121       follower_address: data["followers"],
1122       following_address: data["following"],
1123       bio: data["summary"]
1124     }
1125
1126     # nickname can be nil because of virtual actors
1127     user_data =
1128       if data["preferredUsername"] do
1129         Map.put(
1130           user_data,
1131           :nickname,
1132           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1133         )
1134       else
1135         Map.put(user_data, :nickname, nil)
1136       end
1137
1138     {:ok, user_data}
1139   end
1140
1141   def fetch_follow_information_for_user(user) do
1142     with {:ok, following_data} <-
1143            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1144          following_count when is_integer(following_count) <- following_data["totalItems"],
1145          {:ok, hide_follows} <- collection_private(following_data),
1146          {:ok, followers_data} <-
1147            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1148          followers_count when is_integer(followers_count) <- followers_data["totalItems"],
1149          {:ok, hide_followers} <- collection_private(followers_data) do
1150       {:ok,
1151        %{
1152          hide_follows: hide_follows,
1153          follower_count: followers_count,
1154          following_count: following_count,
1155          hide_followers: hide_followers
1156        }}
1157     else
1158       {:error, _} = e ->
1159         e
1160
1161       e ->
1162         {:error, e}
1163     end
1164   end
1165
1166   defp maybe_update_follow_information(data) do
1167     with {:enabled, true} <-
1168            {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
1169          {:ok, info} <- fetch_follow_information_for_user(data) do
1170       info = Map.merge(data[:info] || %{}, info)
1171       Map.put(data, :info, info)
1172     else
1173       {:enabled, false} ->
1174         data
1175
1176       e ->
1177         Logger.error(
1178           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1179         )
1180
1181         data
1182     end
1183   end
1184
1185   defp collection_private(data) do
1186     if is_map(data["first"]) and
1187          data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
1188       {:ok, false}
1189     else
1190       with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1191              Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
1192         {:ok, false}
1193       else
1194         {:error, {:ok, %{status: code}}} when code in [401, 403] ->
1195           {:ok, true}
1196
1197         {:error, _} = e ->
1198           e
1199
1200         e ->
1201           {:error, e}
1202       end
1203     end
1204   end
1205
1206   def user_data_from_user_object(data) do
1207     with {:ok, data} <- MRF.filter(data),
1208          {:ok, data} <- object_to_user_data(data) do
1209       {:ok, data}
1210     else
1211       e -> {:error, e}
1212     end
1213   end
1214
1215   def fetch_and_prepare_user_from_ap_id(ap_id) do
1216     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1217          {:ok, data} <- user_data_from_user_object(data),
1218          data <- maybe_update_follow_information(data) do
1219       {:ok, data}
1220     else
1221       e ->
1222         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1223         {:error, e}
1224     end
1225   end
1226
1227   def make_user_from_ap_id(ap_id) do
1228     if _user = User.get_cached_by_ap_id(ap_id) do
1229       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1230     else
1231       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1232         User.insert_or_update_user(data)
1233       else
1234         e -> {:error, e}
1235       end
1236     end
1237   end
1238
1239   def make_user_from_nickname(nickname) do
1240     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1241       make_user_from_ap_id(ap_id)
1242     else
1243       _e -> {:error, "No AP id in WebFinger"}
1244     end
1245   end
1246
1247   # filter out broken threads
1248   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1249     entire_thread_visible_for_user?(activity, user)
1250   end
1251
1252   # do post-processing on a specific activity
1253   def contain_activity(%Activity{} = activity, %User{} = user) do
1254     contain_broken_threads(activity, user)
1255   end
1256
1257   def fetch_direct_messages_query do
1258     Activity
1259     |> restrict_type(%{"type" => "Create"})
1260     |> restrict_visibility(%{visibility: "direct"})
1261     |> order_by([activity], asc: activity.id)
1262   end
1263 end