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