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