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