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