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     end
 271   end
 272
 273   def accept(%{to: to, actor: actor, object: object} = params) do
 274     # only accept false as false value
 275     local = !(params[:local] == false)
 276
 277     with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
 278          {:ok, activity} <- insert(data, local),
 279          :ok <- maybe_federate(activity) do
 280       {:ok, activity}
 281     end
 282   end
 283
 284   def reject(%{to: to, actor: actor, object: object} = params) do
 285     # only accept false as false value
 286     local = !(params[:local] == false)
 287
 288     with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
 289          {:ok, activity} <- insert(data, local),
 290          :ok <- maybe_federate(activity) do
 291       {:ok, activity}
 292     end
 293   end
 294
 295   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 296     # only accept false as false value
 297     local = !(params[:local] == false)
 298
 299     with data <- %{
 300            "to" => to,
 301            "cc" => cc,
 302            "type" => "Update",
 303            "actor" => actor,
 304            "object" => object
 305          },
 306          {:ok, activity} <- insert(data, local),
 307          :ok <- maybe_federate(activity) do
 308       {:ok, activity}
 309     end
 310   end
 311
 312   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 313   def like(
 314         %User{ap_id: ap_id} = user,
 315         %Object{data: %{"id" => _}} = object,
 316         activity_id \\ nil,
 317         local \\ true
 318       ) do
 319     with nil <- get_existing_like(ap_id, object),
 320          like_data <- make_like_data(user, object, activity_id),
 321          {:ok, activity} <- insert(like_data, local),
 322          {:ok, object} <- add_like_to_object(activity, object),
 323          :ok <- maybe_federate(activity) do
 324       {:ok, activity, object}
 325     else
 326       %Activity{} = activity -> {:ok, activity, object}
 327       error -> {:error, error}
 328     end
 329   end
 330
 331   def unlike(
 332         %User{} = actor,
 333         %Object{} = object,
 334         activity_id \\ nil,
 335         local \\ true
 336       ) do
 337     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 338          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 339          {:ok, unlike_activity} <- insert(unlike_data, local),
 340          {:ok, _activity} <- Repo.delete(like_activity),
 341          {:ok, object} <- remove_like_from_object(like_activity, object),
 342          :ok <- maybe_federate(unlike_activity) do
 343       {:ok, unlike_activity, like_activity, object}
 344     else
 345       _e -> {:ok, object}
 346     end
 347   end
 348
 349   def announce(
 350         %User{ap_id: _} = user,
 351         %Object{data: %{"id" => _}} = object,
 352         activity_id \\ nil,
 353         local \\ true,
 354         public \\ true
 355       ) do
 356     with true <- is_public?(object),
 357          announce_data <- make_announce_data(user, object, activity_id, public),
 358          {:ok, activity} <- insert(announce_data, local),
 359          {:ok, object} <- add_announce_to_object(activity, object),
 360          :ok <- maybe_federate(activity) do
 361       {:ok, activity, object}
 362     else
 363       error -> {:error, error}
 364     end
 365   end
 366
 367   def unannounce(
 368         %User{} = actor,
 369         %Object{} = object,
 370         activity_id \\ nil,
 371         local \\ true
 372       ) do
 373     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 374          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 375          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 376          :ok <- maybe_federate(unannounce_activity),
 377          {:ok, _activity} <- Repo.delete(announce_activity),
 378          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 379       {:ok, unannounce_activity, object}
 380     else
 381       _e -> {:ok, object}
 382     end
 383   end
 384
 385   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 386     with data <- make_follow_data(follower, followed, activity_id),
 387          {:ok, activity} <- insert(data, local),
 388          :ok <- maybe_federate(activity) 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),
 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     |> restrict_blocked(opts)
 519     |> restrict_recipients(recipients, opts["user"])
 520     |> where(
 521       [activity],
 522       fragment(
 523         "?->>'type' = ? and ?->>'context' = ?",
 524         activity.data,
 525         "Create",
 526         activity.data,
 527         ^context
 528       )
 529     )
 530     |> exclude_poll_votes(opts)
 531     |> order_by([activity], desc: activity.id)
 532   end
 533
 534   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 535   def fetch_activities_for_context(context, opts \\ %{}) do
 536     context
 537     |> fetch_activities_for_context_query(opts)
 538     |> Repo.all()
 539   end
 540
 541   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 542           Pleroma.FlakeId.t() | nil
 543   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 544     context
 545     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 546     |> limit(1)
 547     |> select([a], a.id)
 548     |> Repo.one()
 549   end
 550
 551   def fetch_public_activities(opts \\ %{}) do
 552     q = fetch_activities_query([Pleroma.Constants.as_public()], opts)
 553
 554     q
 555     |> restrict_unlisted()
 556     |> Pagination.fetch_paginated(opts)
 557     |> Enum.reverse()
 558   end
 559
 560   @valid_visibilities ~w[direct unlisted public private]
 561
 562   defp restrict_visibility(query, %{visibility: visibility})
 563        when is_list(visibility) do
 564     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 565       query =
 566         from(
 567           a in query,
 568           where:
 569             fragment(
 570               "activity_visibility(?, ?, ?) = ANY (?)",
 571               a.actor,
 572               a.recipients,
 573               a.data,
 574               ^visibility
 575             )
 576         )
 577
 578       query
 579     else
 580       Logger.error("Could not restrict visibility to #{visibility}")
 581     end
 582   end
 583
 584   defp restrict_visibility(query, %{visibility: visibility})
 585        when visibility in @valid_visibilities do
 586     from(
 587       a in query,
 588       where:
 589         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 590     )
 591   end
 592
 593   defp restrict_visibility(_query, %{visibility: visibility})
 594        when visibility not in @valid_visibilities do
 595     Logger.error("Could not restrict visibility to #{visibility}")
 596   end
 597
 598   defp restrict_visibility(query, _visibility), do: query
 599
 600   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 601     do: query
 602
 603   defp restrict_thread_visibility(
 604          query,
 605          %{"user" => %User{info: %{skip_thread_containment: true}}},
 606          _
 607        ),
 608        do: query
 609
 610   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 611     from(
 612       a in query,
 613       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 614     )
 615   end
 616
 617   defp restrict_thread_visibility(query, _, _), do: query
 618
 619   def fetch_user_activities(user, reading_user, params \\ %{}) do
 620     params =
 621       params
 622       |> Map.put("type", ["Create", "Announce"])
 623       |> Map.put("actor_id", user.ap_id)
 624       |> Map.put("whole_db", true)
 625       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 626
 627     recipients =
 628       user_activities_recipients(%{
 629         "godmode" => params["godmode"],
 630         "reading_user" => reading_user
 631       })
 632
 633     fetch_activities(recipients, params)
 634     |> Enum.reverse()
 635   end
 636
 637   defp user_activities_recipients(%{"godmode" => true}) do
 638     []
 639   end
 640
 641   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 642     if reading_user do
 643       [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
 644     else
 645       [Pleroma.Constants.as_public()]
 646     end
 647   end
 648
 649   defp restrict_since(query, %{"since_id" => ""}), do: query
 650
 651   defp restrict_since(query, %{"since_id" => since_id}) do
 652     from(activity in query, where: activity.id > ^since_id)
 653   end
 654
 655   defp restrict_since(query, _), do: query
 656
 657   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 658     raise "Can't use the child object without preloading!"
 659   end
 660
 661   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 662        when is_list(tag_reject) and tag_reject != [] do
 663     from(
 664       [_activity, object] in query,
 665       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 666     )
 667   end
 668
 669   defp restrict_tag_reject(query, _), do: query
 670
 671   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 672     raise "Can't use the child object without preloading!"
 673   end
 674
 675   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 676        when is_list(tag_all) and tag_all != [] do
 677     from(
 678       [_activity, object] in query,
 679       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 680     )
 681   end
 682
 683   defp restrict_tag_all(query, _), do: query
 684
 685   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 686     raise "Can't use the child object without preloading!"
 687   end
 688
 689   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 690     from(
 691       [_activity, object] in query,
 692       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 693     )
 694   end
 695
 696   defp restrict_tag(query, %{"tag" => tag}) when is_binary(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, _), do: query
 704
 705   defp restrict_recipients(query, [], _user), do: query
 706
 707   defp restrict_recipients(query, recipients, nil) do
 708     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 709   end
 710
 711   defp restrict_recipients(query, recipients, user) do
 712     from(
 713       activity in query,
 714       where: fragment("? && ?", ^recipients, activity.recipients),
 715       or_where: activity.actor == ^user.ap_id
 716     )
 717   end
 718
 719   defp restrict_local(query, %{"local_only" => true}) do
 720     from(activity in query, where: activity.local == true)
 721   end
 722
 723   defp restrict_local(query, _), do: query
 724
 725   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 726     from(activity in query, where: activity.actor == ^actor_id)
 727   end
 728
 729   defp restrict_actor(query, _), do: query
 730
 731   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 732     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 733   end
 734
 735   defp restrict_type(query, %{"type" => type}) do
 736     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 737   end
 738
 739   defp restrict_type(query, _), do: query
 740
 741   defp restrict_state(query, %{"state" => state}) do
 742     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 743   end
 744
 745   defp restrict_state(query, _), do: query
 746
 747   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 748     from(
 749       activity in query,
 750       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 751     )
 752   end
 753
 754   defp restrict_favorited_by(query, _), do: query
 755
 756   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 757     raise "Can't use the child object without preloading!"
 758   end
 759
 760   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 761     from(
 762       [_activity, object] in query,
 763       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 764     )
 765   end
 766
 767   defp restrict_media(query, _), do: query
 768
 769   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 770     from(
 771       activity in query,
 772       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 773     )
 774   end
 775
 776   defp restrict_replies(query, _), do: query
 777
 778   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 779     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 780   end
 781
 782   defp restrict_reblogs(query, _), do: query
 783
 784   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 785
 786   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 787     mutes = info.mutes
 788
 789     from(
 790       activity in query,
 791       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 792       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 793     )
 794   end
 795
 796   defp restrict_muted(query, _), do: query
 797
 798   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 799     blocks = info.blocks || []
 800     domain_blocks = info.domain_blocks || []
 801
 802     query =
 803       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 804
 805     from(
 806       [activity, object: o] in query,
 807       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 808       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 809       where:
 810         fragment(
 811           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 812           activity.data,
 813           activity.data,
 814           ^blocks
 815         ),
 816       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 817       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 818     )
 819   end
 820
 821   defp restrict_blocked(query, _), do: query
 822
 823   defp restrict_unlisted(query) do
 824     from(
 825       activity in query,
 826       where:
 827         fragment(
 828           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 829           activity.data,
 830           ^[Pleroma.Constants.as_public()]
 831         )
 832     )
 833   end
 834
 835   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 836     from(activity in query, where: activity.id in ^ids)
 837   end
 838
 839   defp restrict_pinned(query, _), do: query
 840
 841   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 842     muted_reblogs = info.muted_reblogs || []
 843
 844     from(
 845       activity in query,
 846       where:
 847         fragment(
 848           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 849           activity.data,
 850           activity.actor,
 851           ^muted_reblogs
 852         )
 853     )
 854   end
 855
 856   defp restrict_muted_reblogs(query, _), do: query
 857
 858   defp exclude_poll_votes(query, %{"include_poll_votes" => "true"}), do: query
 859
 860   defp exclude_poll_votes(query, _) do
 861     if has_named_binding?(query, :object) do
 862       from([activity, object: o] in query,
 863         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 864       )
 865     else
 866       query
 867     end
 868   end
 869
 870   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 871
 872   defp maybe_preload_objects(query, _) do
 873     query
 874     |> Activity.with_preloaded_object()
 875   end
 876
 877   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 878
 879   defp maybe_preload_bookmarks(query, opts) do
 880     query
 881     |> Activity.with_preloaded_bookmark(opts["user"])
 882   end
 883
 884   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 885
 886   defp maybe_set_thread_muted_field(query, opts) do
 887     query
 888     |> Activity.with_set_thread_muted_field(opts["user"])
 889   end
 890
 891   defp maybe_order(query, %{order: :desc}) do
 892     query
 893     |> order_by(desc: :id)
 894   end
 895
 896   defp maybe_order(query, %{order: :asc}) do
 897     query
 898     |> order_by(asc: :id)
 899   end
 900
 901   defp maybe_order(query, _), do: query
 902
 903   def fetch_activities_query(recipients, opts \\ %{}) do
 904     config = %{
 905       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
 906     }
 907
 908     Activity
 909     |> maybe_preload_objects(opts)
 910     |> maybe_preload_bookmarks(opts)
 911     |> maybe_set_thread_muted_field(opts)
 912     |> maybe_order(opts)
 913     |> restrict_recipients(recipients, opts["user"])
 914     |> restrict_tag(opts)
 915     |> restrict_tag_reject(opts)
 916     |> restrict_tag_all(opts)
 917     |> restrict_since(opts)
 918     |> restrict_local(opts)
 919     |> restrict_actor(opts)
 920     |> restrict_type(opts)
 921     |> restrict_state(opts)
 922     |> restrict_favorited_by(opts)
 923     |> restrict_blocked(opts)
 924     |> restrict_muted(opts)
 925     |> restrict_media(opts)
 926     |> restrict_visibility(opts)
 927     |> restrict_thread_visibility(opts, config)
 928     |> restrict_replies(opts)
 929     |> restrict_reblogs(opts)
 930     |> restrict_pinned(opts)
 931     |> restrict_muted_reblogs(opts)
 932     |> Activity.restrict_deactivated_users()
 933     |> exclude_poll_votes(opts)
 934   end
 935
 936   def fetch_activities(recipients, opts \\ %{}) do
 937     list_memberships = Pleroma.List.memberships(opts["user"])
 938
 939     fetch_activities_query(recipients ++ list_memberships, opts)
 940     |> Pagination.fetch_paginated(opts)
 941     |> Enum.reverse()
 942     |> maybe_update_cc(list_memberships, opts["user"])
 943   end
 944
 945   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
 946        when is_list(list_memberships) and length(list_memberships) > 0 do
 947     Enum.map(activities, fn
 948       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
 949         if Enum.any?(bcc, &(&1 in list_memberships)) do
 950           update_in(activity.data["cc"], &[user_ap_id | &1])
 951         else
 952           activity
 953         end
 954
 955       activity ->
 956         activity
 957     end)
 958   end
 959
 960   defp maybe_update_cc(activities, _, _), do: activities
 961
 962   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
 963     from(activity in query,
 964       where:
 965         fragment("? && ?", activity.recipients, ^recipients) or
 966           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
 967              ^Pleroma.Constants.as_public() in activity.recipients)
 968     )
 969   end
 970
 971   def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
 972     fetch_activities_query([], opts)
 973     |> fetch_activities_bounded_query(recipients, recipients_with_public)
 974     |> Pagination.fetch_paginated(opts)
 975     |> Enum.reverse()
 976   end
 977
 978   def upload(file, opts \\ []) do
 979     with {:ok, data} <- Upload.store(file, opts) do
 980       obj_data =
 981         if opts[:actor] do
 982           Map.put(data, "actor", opts[:actor])
 983         else
 984           data
 985         end
 986
 987       Repo.insert(%Object{data: obj_data})
 988     end
 989   end
 990
 991   defp object_to_user_data(data) do
 992     avatar =
 993       data["icon"]["url"] &&
 994         %{
 995           "type" => "Image",
 996           "url" => [%{"href" => data["icon"]["url"]}]
 997         }
 998
 999     banner =
1000       data["image"]["url"] &&
1001         %{
1002           "type" => "Image",
1003           "url" => [%{"href" => data["image"]["url"]}]
1004         }
1005
1006     locked = data["manuallyApprovesFollowers"] || false
1007     data = Transmogrifier.maybe_fix_user_object(data)
1008
1009     user_data = %{
1010       ap_id: data["id"],
1011       info: %{
1012         "ap_enabled" => true,
1013         "source_data" => data,
1014         "banner" => banner,
1015         "locked" => locked
1016       },
1017       avatar: avatar,
1018       name: data["name"],
1019       follower_address: data["followers"],
1020       following_address: data["following"],
1021       bio: data["summary"]
1022     }
1023
1024     # nickname can be nil because of virtual actors
1025     user_data =
1026       if data["preferredUsername"] do
1027         Map.put(
1028           user_data,
1029           :nickname,
1030           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1031         )
1032       else
1033         Map.put(user_data, :nickname, nil)
1034       end
1035
1036     {:ok, user_data}
1037   end
1038
1039   def user_data_from_user_object(data) do
1040     with {:ok, data} <- MRF.filter(data),
1041          {:ok, data} <- object_to_user_data(data) do
1042       {:ok, data}
1043     else
1044       e -> {:error, e}
1045     end
1046   end
1047
1048   def fetch_and_prepare_user_from_ap_id(ap_id) do
1049     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1050          {:ok, data} <- user_data_from_user_object(data) do
1051       {:ok, data}
1052     else
1053       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1054     end
1055   end
1056
1057   def make_user_from_ap_id(ap_id) do
1058     if _user = User.get_cached_by_ap_id(ap_id) do
1059       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1060     else
1061       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1062         User.insert_or_update_user(data)
1063       else
1064         e -> {:error, e}
1065       end
1066     end
1067   end
1068
1069   def make_user_from_nickname(nickname) do
1070     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1071       make_user_from_ap_id(ap_id)
1072     else
1073       _e -> {:error, "No AP id in WebFinger"}
1074     end
1075   end
1076
1077   # filter out broken threads
1078   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1079     entire_thread_visible_for_user?(activity, user)
1080   end
1081
1082   # do post-processing on a specific activity
1083   def contain_activity(%Activity{} = activity, %User{} = user) do
1084     contain_broken_threads(activity, user)
1085   end
1086
1087   def fetch_direct_messages_query do
1088     Activity
1089     |> restrict_type(%{"type" => "Create"})
1090     |> restrict_visibility(%{visibility: "direct"})
1091     |> order_by([activity], asc: activity.id)
1092   end
1093 end