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     from(
 754       activity in query,
 755       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 756       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 757       where:
 758         fragment(
 759           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 760           activity.data,
 761           activity.data,
 762           ^blocks
 763         ),
 764       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
 765     )
 766   end
 767
 768   defp restrict_blocked(query, _), do: query
 769
 770   defp restrict_unlisted(query) do
 771     from(
 772       activity in query,
 773       where:
 774         fragment(
 775           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 776           activity.data,
 777           ^["https://www.w3.org/ns/activitystreams#Public"]
 778         )
 779     )
 780   end
 781
 782   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 783     from(activity in query, where: activity.id in ^ids)
 784   end
 785
 786   defp restrict_pinned(query, _), do: query
 787
 788   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 789     muted_reblogs = info.muted_reblogs || []
 790
 791     from(
 792       activity in query,
 793       where:
 794         fragment(
 795           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 796           activity.data,
 797           activity.actor,
 798           ^muted_reblogs
 799         )
 800     )
 801   end
 802
 803   defp restrict_muted_reblogs(query, _), do: query
 804
 805   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 806
 807   defp maybe_preload_objects(query, _) do
 808     query
 809     |> Activity.with_preloaded_object()
 810   end
 811
 812   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 813
 814   defp maybe_preload_bookmarks(query, opts) do
 815     query
 816     |> Activity.with_preloaded_bookmark(opts["user"])
 817   end
 818
 819   defp maybe_order(query, %{order: :desc}) do
 820     query
 821     |> order_by(desc: :id)
 822   end
 823
 824   defp maybe_order(query, %{order: :asc}) do
 825     query
 826     |> order_by(asc: :id)
 827   end
 828
 829   defp maybe_order(query, _), do: query
 830
 831   def fetch_activities_query(recipients, opts \\ %{}) do
 832     base_query = from(activity in Activity)
 833
 834     base_query
 835     |> maybe_preload_objects(opts)
 836     |> maybe_preload_bookmarks(opts)
 837     |> maybe_order(opts)
 838     |> restrict_recipients(recipients, opts["user"])
 839     |> restrict_tag(opts)
 840     |> restrict_tag_reject(opts)
 841     |> restrict_tag_all(opts)
 842     |> restrict_since(opts)
 843     |> restrict_local(opts)
 844     |> restrict_actor(opts)
 845     |> restrict_type(opts)
 846     |> restrict_favorited_by(opts)
 847     |> restrict_blocked(opts)
 848     |> restrict_muted(opts)
 849     |> restrict_media(opts)
 850     |> restrict_visibility(opts)
 851     |> restrict_replies(opts)
 852     |> restrict_reblogs(opts)
 853     |> restrict_pinned(opts)
 854     |> restrict_muted_reblogs(opts)
 855   end
 856
 857   def fetch_activities(recipients, opts \\ %{}) do
 858     fetch_activities_query(recipients, opts)
 859     |> Pagination.fetch_paginated(opts)
 860     |> Enum.reverse()
 861   end
 862
 863   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 864     fetch_activities_query([], opts)
 865     |> restrict_to_cc(recipients_to, recipients_cc)
 866     |> Pagination.fetch_paginated(opts)
 867     |> Enum.reverse()
 868   end
 869
 870   def upload(file, opts \\ []) do
 871     with {:ok, data} <- Upload.store(file, opts) do
 872       obj_data =
 873         if opts[:actor] do
 874           Map.put(data, "actor", opts[:actor])
 875         else
 876           data
 877         end
 878
 879       Repo.insert(%Object{data: obj_data})
 880     end
 881   end
 882
 883   def user_data_from_user_object(data) do
 884     avatar =
 885       data["icon"]["url"] &&
 886         %{
 887           "type" => "Image",
 888           "url" => [%{"href" => data["icon"]["url"]}]
 889         }
 890
 891     banner =
 892       data["image"]["url"] &&
 893         %{
 894           "type" => "Image",
 895           "url" => [%{"href" => data["image"]["url"]}]
 896         }
 897
 898     locked = data["manuallyApprovesFollowers"] || false
 899     data = Transmogrifier.maybe_fix_user_object(data)
 900
 901     user_data = %{
 902       ap_id: data["id"],
 903       info: %{
 904         "ap_enabled" => true,
 905         "source_data" => data,
 906         "banner" => banner,
 907         "locked" => locked
 908       },
 909       avatar: avatar,
 910       name: data["name"],
 911       follower_address: data["followers"],
 912       bio: data["summary"]
 913     }
 914
 915     # nickname can be nil because of virtual actors
 916     user_data =
 917       if data["preferredUsername"] do
 918         Map.put(
 919           user_data,
 920           :nickname,
 921           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 922         )
 923       else
 924         Map.put(user_data, :nickname, nil)
 925       end
 926
 927     {:ok, user_data}
 928   end
 929
 930   def fetch_and_prepare_user_from_ap_id(ap_id) do
 931     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
 932       user_data_from_user_object(data)
 933     else
 934       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 935     end
 936   end
 937
 938   def make_user_from_ap_id(ap_id) do
 939     if _user = User.get_cached_by_ap_id(ap_id) do
 940       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 941     else
 942       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 943         User.insert_or_update_user(data)
 944       else
 945         e -> {:error, e}
 946       end
 947     end
 948   end
 949
 950   def make_user_from_nickname(nickname) do
 951     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 952       make_user_from_ap_id(ap_id)
 953     else
 954       _e -> {:error, "No AP id in WebFinger"}
 955     end
 956   end
 957
 958   # filter out broken threads
 959   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 960     entire_thread_visible_for_user?(activity, user)
 961   end
 962
 963   # do post-processing on a specific activity
 964   def contain_activity(%Activity{} = activity, %User{} = user) do
 965     contain_broken_threads(activity, user)
 966   end
 967
 968   # do post-processing on a timeline
 969   def contain_timeline(timeline, user) do
 970     timeline
 971     |> Enum.filter(fn activity ->
 972       contain_activity(activity, user)
 973     end)
 974   end
 975 end