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