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