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