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     outgoing_blocks = Pleroma.Config.get([:activitypub, :outgoing_blocks])
 403     unfollow_blocked = Pleroma.Config.get([:activitypub, :unfollow_blocked])
 404
 405     if unfollow_blocked do
 406       follow_activity = fetch_latest_follow(blocker, blocked)
 407       if follow_activity, do: unfollow(blocker, blocked, nil, local)
 408     end
 409
 410     with true <- outgoing_blocks,
 411          block_data <- make_block_data(blocker, blocked, activity_id),
 412          {:ok, activity} <- insert(block_data, local),
 413          :ok <- maybe_federate(activity) do
 414       {:ok, activity}
 415     else
 416       _e -> {:ok, nil}
 417     end
 418   end
 419
 420   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 421     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 422          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 423          {:ok, activity} <- insert(unblock_data, local),
 424          :ok <- maybe_federate(activity) do
 425       {:ok, activity}
 426     end
 427   end
 428
 429   def flag(
 430         %{
 431           actor: actor,
 432           context: context,
 433           account: account,
 434           statuses: statuses,
 435           content: content
 436         } = params
 437       ) do
 438     # only accept false as false value
 439     local = !(params[:local] == false)
 440     forward = !(params[:forward] == false)
 441
 442     additional = params[:additional] || %{}
 443
 444     params = %{
 445       actor: actor,
 446       context: context,
 447       account: account,
 448       statuses: statuses,
 449       content: content
 450     }
 451
 452     additional =
 453       if forward do
 454         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 455       else
 456         Map.merge(additional, %{"to" => [], "cc" => []})
 457       end
 458
 459     with flag_data <- make_flag_data(params, additional),
 460          {:ok, activity} <- insert(flag_data, local),
 461          :ok <- maybe_federate(activity) do
 462       Enum.each(User.all_superusers(), fn superuser ->
 463         superuser
 464         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 465         |> Pleroma.Emails.Mailer.deliver_async()
 466       end)
 467
 468       {:ok, activity}
 469     end
 470   end
 471
 472   defp fetch_activities_for_context_query(context, opts) do
 473     public = ["https://www.w3.org/ns/activitystreams#Public"]
 474
 475     recipients =
 476       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 477
 478     from(activity in Activity)
 479     |> restrict_blocked(opts)
 480     |> restrict_recipients(recipients, opts["user"])
 481     |> where(
 482       [activity],
 483       fragment(
 484         "?->>'type' = ? and ?->>'context' = ?",
 485         activity.data,
 486         "Create",
 487         activity.data,
 488         ^context
 489       )
 490     )
 491     |> order_by([activity], desc: activity.id)
 492   end
 493
 494   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 495   def fetch_activities_for_context(context, opts \\ %{}) do
 496     context
 497     |> fetch_activities_for_context_query(opts)
 498     |> Activity.with_preloaded_object()
 499     |> Repo.all()
 500   end
 501
 502   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 503           Pleroma.FlakeId.t() | nil
 504   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 505     context
 506     |> fetch_activities_for_context_query(opts)
 507     |> limit(1)
 508     |> select([a], a.id)
 509     |> Repo.one()
 510   end
 511
 512   def fetch_public_activities(opts \\ %{}) do
 513     q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
 514
 515     q
 516     |> restrict_unlisted()
 517     |> Pagination.fetch_paginated(opts)
 518     |> Enum.reverse()
 519   end
 520
 521   @valid_visibilities ~w[direct unlisted public private]
 522
 523   defp restrict_visibility(query, %{visibility: visibility})
 524        when is_list(visibility) do
 525     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 526       query =
 527         from(
 528           a in query,
 529           where:
 530             fragment(
 531               "activity_visibility(?, ?, ?) = ANY (?)",
 532               a.actor,
 533               a.recipients,
 534               a.data,
 535               ^visibility
 536             )
 537         )
 538
 539       query
 540     else
 541       Logger.error("Could not restrict visibility to #{visibility}")
 542     end
 543   end
 544
 545   defp restrict_visibility(query, %{visibility: visibility})
 546        when visibility in @valid_visibilities do
 547     query =
 548       from(
 549         a in query,
 550         where:
 551           fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 552       )
 553
 554     query
 555   end
 556
 557   defp restrict_visibility(_query, %{visibility: visibility})
 558        when visibility not in @valid_visibilities do
 559     Logger.error("Could not restrict visibility to #{visibility}")
 560   end
 561
 562   defp restrict_visibility(query, _visibility), do: query
 563
 564   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}) do
 565     query =
 566       from(
 567         a in query,
 568         where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 569       )
 570
 571     query
 572   end
 573
 574   defp restrict_thread_visibility(query, _), do: query
 575
 576   def fetch_user_activities(user, reading_user, params \\ %{}) do
 577     params =
 578       params
 579       |> Map.put("type", ["Create", "Announce"])
 580       |> Map.put("actor_id", user.ap_id)
 581       |> Map.put("whole_db", true)
 582       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 583
 584     recipients =
 585       if reading_user do
 586         ["https://www.w3.org/ns/activitystreams#Public"] ++
 587           [reading_user.ap_id | reading_user.following]
 588       else
 589         ["https://www.w3.org/ns/activitystreams#Public"]
 590       end
 591
 592     fetch_activities(recipients, params)
 593     |> Enum.reverse()
 594   end
 595
 596   defp restrict_since(query, %{"since_id" => ""}), do: query
 597
 598   defp restrict_since(query, %{"since_id" => since_id}) do
 599     from(activity in query, where: activity.id > ^since_id)
 600   end
 601
 602   defp restrict_since(query, _), do: query
 603
 604   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 605     raise "Can't use the child object without preloading!"
 606   end
 607
 608   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 609        when is_list(tag_reject) and tag_reject != [] do
 610     from(
 611       [_activity, object] in query,
 612       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 613     )
 614   end
 615
 616   defp restrict_tag_reject(query, _), do: query
 617
 618   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 619     raise "Can't use the child object without preloading!"
 620   end
 621
 622   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 623        when is_list(tag_all) and tag_all != [] do
 624     from(
 625       [_activity, object] in query,
 626       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 627     )
 628   end
 629
 630   defp restrict_tag_all(query, _), do: query
 631
 632   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 633     raise "Can't use the child object without preloading!"
 634   end
 635
 636   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 637     from(
 638       [_activity, object] in query,
 639       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 640     )
 641   end
 642
 643   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 644     from(
 645       [_activity, object] in query,
 646       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 647     )
 648   end
 649
 650   defp restrict_tag(query, _), do: query
 651
 652   defp restrict_recipients(query, [], _user), do: query
 653
 654   defp restrict_recipients(query, recipients, nil) do
 655     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 656   end
 657
 658   defp restrict_recipients(query, recipients, user) do
 659     from(
 660       activity in query,
 661       where: fragment("? && ?", ^recipients, activity.recipients),
 662       or_where: activity.actor == ^user.ap_id
 663     )
 664   end
 665
 666   defp restrict_local(query, %{"local_only" => true}) do
 667     from(activity in query, where: activity.local == true)
 668   end
 669
 670   defp restrict_local(query, _), do: query
 671
 672   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 673     from(activity in query, where: activity.actor == ^actor_id)
 674   end
 675
 676   defp restrict_actor(query, _), do: query
 677
 678   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 679     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 680   end
 681
 682   defp restrict_type(query, %{"type" => type}) do
 683     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 684   end
 685
 686   defp restrict_type(query, _), do: query
 687
 688   defp restrict_state(query, %{"state" => state}) do
 689     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 690   end
 691
 692   defp restrict_state(query, _), do: query
 693
 694   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 695     from(
 696       activity in query,
 697       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 698     )
 699   end
 700
 701   defp restrict_favorited_by(query, _), do: query
 702
 703   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 704     raise "Can't use the child object without preloading!"
 705   end
 706
 707   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 708     from(
 709       [_activity, object] in query,
 710       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 711     )
 712   end
 713
 714   defp restrict_media(query, _), do: query
 715
 716   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 717     from(
 718       activity in query,
 719       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 720     )
 721   end
 722
 723   defp restrict_replies(query, _), do: query
 724
 725   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 726     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 727   end
 728
 729   defp restrict_reblogs(query, _), do: query
 730
 731   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 732
 733   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 734     mutes = info.mutes
 735
 736     from(
 737       activity in query,
 738       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 739       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 740     )
 741   end
 742
 743   defp restrict_muted(query, _), do: query
 744
 745   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 746     blocks = info.blocks || []
 747     domain_blocks = info.domain_blocks || []
 748
 749     query =
 750       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 751
 752     from(
 753       [activity, object: o] in query,
 754       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 755       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 756       where:
 757         fragment(
 758           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 759           activity.data,
 760           activity.data,
 761           ^blocks
 762         ),
 763       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 764       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^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_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 820
 821   defp maybe_set_thread_muted_field(query, opts) do
 822     query
 823     |> Activity.with_set_thread_muted_field(opts["user"])
 824   end
 825
 826   defp maybe_order(query, %{order: :desc}) do
 827     query
 828     |> order_by(desc: :id)
 829   end
 830
 831   defp maybe_order(query, %{order: :asc}) do
 832     query
 833     |> order_by(asc: :id)
 834   end
 835
 836   defp maybe_order(query, _), do: query
 837
 838   def fetch_activities_query(recipients, opts \\ %{}) do
 839     base_query = from(activity in Activity)
 840
 841     base_query
 842     |> maybe_preload_objects(opts)
 843     |> maybe_preload_bookmarks(opts)
 844     |> maybe_set_thread_muted_field(opts)
 845     |> maybe_order(opts)
 846     |> restrict_recipients(recipients, opts["user"])
 847     |> restrict_tag(opts)
 848     |> restrict_tag_reject(opts)
 849     |> restrict_tag_all(opts)
 850     |> restrict_since(opts)
 851     |> restrict_local(opts)
 852     |> restrict_actor(opts)
 853     |> restrict_type(opts)
 854     |> restrict_state(opts)
 855     |> restrict_favorited_by(opts)
 856     |> restrict_blocked(opts)
 857     |> restrict_muted(opts)
 858     |> restrict_media(opts)
 859     |> restrict_visibility(opts)
 860     |> restrict_thread_visibility(opts)
 861     |> restrict_replies(opts)
 862     |> restrict_reblogs(opts)
 863     |> restrict_pinned(opts)
 864     |> restrict_muted_reblogs(opts)
 865     |> Activity.restrict_deactivated_users()
 866   end
 867
 868   def fetch_activities(recipients, opts \\ %{}) do
 869     fetch_activities_query(recipients, opts)
 870     |> Pagination.fetch_paginated(opts)
 871     |> Enum.reverse()
 872   end
 873
 874   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
 875     from(activity in query,
 876       where:
 877         fragment("? && ?", activity.recipients, ^recipients) or
 878           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
 879              "https://www.w3.org/ns/activitystreams#Public" in activity.recipients)
 880     )
 881   end
 882
 883   def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
 884     fetch_activities_query([], opts)
 885     |> fetch_activities_bounded_query(recipients, recipients_with_public)
 886     |> Pagination.fetch_paginated(opts)
 887     |> Enum.reverse()
 888   end
 889
 890   def upload(file, opts \\ []) do
 891     with {:ok, data} <- Upload.store(file, opts) do
 892       obj_data =
 893         if opts[:actor] do
 894           Map.put(data, "actor", opts[:actor])
 895         else
 896           data
 897         end
 898
 899       Repo.insert(%Object{data: obj_data})
 900     end
 901   end
 902
 903   defp object_to_user_data(data) do
 904     avatar =
 905       data["icon"]["url"] &&
 906         %{
 907           "type" => "Image",
 908           "url" => [%{"href" => data["icon"]["url"]}]
 909         }
 910
 911     banner =
 912       data["image"]["url"] &&
 913         %{
 914           "type" => "Image",
 915           "url" => [%{"href" => data["image"]["url"]}]
 916         }
 917
 918     locked = data["manuallyApprovesFollowers"] || false
 919     data = Transmogrifier.maybe_fix_user_object(data)
 920
 921     user_data = %{
 922       ap_id: data["id"],
 923       info: %{
 924         "ap_enabled" => true,
 925         "source_data" => data,
 926         "banner" => banner,
 927         "locked" => locked
 928       },
 929       avatar: avatar,
 930       name: data["name"],
 931       follower_address: data["followers"],
 932       bio: data["summary"]
 933     }
 934
 935     # nickname can be nil because of virtual actors
 936     user_data =
 937       if data["preferredUsername"] do
 938         Map.put(
 939           user_data,
 940           :nickname,
 941           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 942         )
 943       else
 944         Map.put(user_data, :nickname, nil)
 945       end
 946
 947     {:ok, user_data}
 948   end
 949
 950   def user_data_from_user_object(data) do
 951     with {:ok, data} <- MRF.filter(data),
 952          {:ok, data} <- object_to_user_data(data) do
 953       {:ok, data}
 954     else
 955       e -> {:error, e}
 956     end
 957   end
 958
 959   def fetch_and_prepare_user_from_ap_id(ap_id) do
 960     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
 961          {:ok, data} <- user_data_from_user_object(data) do
 962       {:ok, data}
 963     else
 964       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 965     end
 966   end
 967
 968   def make_user_from_ap_id(ap_id) do
 969     if _user = User.get_cached_by_ap_id(ap_id) do
 970       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 971     else
 972       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 973         User.insert_or_update_user(data)
 974       else
 975         e -> {:error, e}
 976       end
 977     end
 978   end
 979
 980   def make_user_from_nickname(nickname) do
 981     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 982       make_user_from_ap_id(ap_id)
 983     else
 984       _e -> {:error, "No AP id in WebFinger"}
 985     end
 986   end
 987
 988   # filter out broken threads
 989   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 990     entire_thread_visible_for_user?(activity, user)
 991   end
 992
 993   # do post-processing on a specific activity
 994   def contain_activity(%Activity{} = activity, %User{} = user) do
 995     contain_broken_threads(activity, user)
 996   end
 997
 998   def fetch_direct_messages_query do
 999     Activity
1000     |> restrict_type(%{"type" => "Create"})
1001     |> restrict_visibility(%{visibility: "direct"})
1002     |> order_by([activity], asc: activity.id)
1003   end
1004 end