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