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       query
 544     else
 545       Logger.error("Could not restrict visibility to #{visibility}")
 546     end
 547   end
 548
 549   defp restrict_visibility(query, %{visibility: visibility})
 550        when visibility in @valid_visibilities do
 551     query =
 552       from(
 553         a in query,
 554         where:
 555           fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 556       )
 557
 558     query
 559   end
 560
 561   defp restrict_visibility(_query, %{visibility: visibility})
 562        when visibility not in @valid_visibilities do
 563     Logger.error("Could not restrict visibility to #{visibility}")
 564   end
 565
 566   defp restrict_visibility(query, _visibility), do: query
 567
 568   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}) do
 569     query =
 570       from(
 571         a in query,
 572         where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 573       )
 574
 575     query
 576   end
 577
 578   defp restrict_thread_visibility(query, _), do: query
 579
 580   def fetch_user_activities(user, reading_user, params \\ %{}) do
 581     params =
 582       params
 583       |> Map.put("type", ["Create", "Announce"])
 584       |> Map.put("actor_id", user.ap_id)
 585       |> Map.put("whole_db", true)
 586       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 587
 588     recipients =
 589       if reading_user do
 590         ["https://www.w3.org/ns/activitystreams#Public"] ++
 591           [reading_user.ap_id | reading_user.following]
 592       else
 593         ["https://www.w3.org/ns/activitystreams#Public"]
 594       end
 595
 596     fetch_activities(recipients, params)
 597     |> Enum.reverse()
 598   end
 599
 600   defp restrict_since(query, %{"since_id" => ""}), do: query
 601
 602   defp restrict_since(query, %{"since_id" => since_id}) do
 603     from(activity in query, where: activity.id > ^since_id)
 604   end
 605
 606   defp restrict_since(query, _), do: query
 607
 608   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 609     raise "Can't use the child object without preloading!"
 610   end
 611
 612   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 613        when is_list(tag_reject) and tag_reject != [] do
 614     from(
 615       [_activity, object] in query,
 616       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 617     )
 618   end
 619
 620   defp restrict_tag_reject(query, _), do: query
 621
 622   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 623     raise "Can't use the child object without preloading!"
 624   end
 625
 626   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 627        when is_list(tag_all) and tag_all != [] do
 628     from(
 629       [_activity, object] in query,
 630       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 631     )
 632   end
 633
 634   defp restrict_tag_all(query, _), do: query
 635
 636   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 637     raise "Can't use the child object without preloading!"
 638   end
 639
 640   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 641     from(
 642       [_activity, object] in query,
 643       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 644     )
 645   end
 646
 647   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 648     from(
 649       [_activity, object] in query,
 650       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 651     )
 652   end
 653
 654   defp restrict_tag(query, _), do: query
 655
 656   defp restrict_to_cc(query, recipients_to, recipients_cc) do
 657     from(
 658       activity in query,
 659       where:
 660         fragment(
 661           "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
 662           activity.data,
 663           ^recipients_to,
 664           activity.data,
 665           ^recipients_cc
 666         )
 667     )
 668   end
 669
 670   defp restrict_recipients(query, [], _user), do: query
 671
 672   defp restrict_recipients(query, recipients, nil) do
 673     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 674   end
 675
 676   defp restrict_recipients(query, recipients, user) do
 677     from(
 678       activity in query,
 679       where: fragment("? && ?", ^recipients, activity.recipients),
 680       or_where: activity.actor == ^user.ap_id
 681     )
 682   end
 683
 684   defp restrict_local(query, %{"local_only" => true}) do
 685     from(activity in query, where: activity.local == true)
 686   end
 687
 688   defp restrict_local(query, _), do: query
 689
 690   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 691     from(activity in query, where: activity.actor == ^actor_id)
 692   end
 693
 694   defp restrict_actor(query, _), do: query
 695
 696   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 697     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 698   end
 699
 700   defp restrict_type(query, %{"type" => type}) do
 701     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 702   end
 703
 704   defp restrict_type(query, _), do: query
 705
 706   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 707     from(
 708       activity in query,
 709       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 710     )
 711   end
 712
 713   defp restrict_favorited_by(query, _), do: query
 714
 715   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 716     raise "Can't use the child object without preloading!"
 717   end
 718
 719   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 720     from(
 721       [_activity, object] in query,
 722       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 723     )
 724   end
 725
 726   defp restrict_media(query, _), do: query
 727
 728   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 729     from(
 730       activity in query,
 731       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 732     )
 733   end
 734
 735   defp restrict_replies(query, _), do: query
 736
 737   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 738     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 739   end
 740
 741   defp restrict_reblogs(query, _), do: query
 742
 743   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 744
 745   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 746     mutes = info.mutes
 747
 748     from(
 749       activity in query,
 750       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 751       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 752     )
 753   end
 754
 755   defp restrict_muted(query, _), do: query
 756
 757   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 758     blocks = info.blocks || []
 759     domain_blocks = info.domain_blocks || []
 760
 761     from(
 762       activity in query,
 763       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 764       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 765       where:
 766         fragment(
 767           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 768           activity.data,
 769           activity.data,
 770           ^blocks
 771         ),
 772       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
 773     )
 774   end
 775
 776   defp restrict_blocked(query, _), do: query
 777
 778   defp restrict_unlisted(query) do
 779     from(
 780       activity in query,
 781       where:
 782         fragment(
 783           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 784           activity.data,
 785           ^["https://www.w3.org/ns/activitystreams#Public"]
 786         )
 787     )
 788   end
 789
 790   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 791     from(activity in query, where: activity.id in ^ids)
 792   end
 793
 794   defp restrict_pinned(query, _), do: query
 795
 796   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 797     muted_reblogs = info.muted_reblogs || []
 798
 799     from(
 800       activity in query,
 801       where:
 802         fragment(
 803           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 804           activity.data,
 805           activity.actor,
 806           ^muted_reblogs
 807         )
 808     )
 809   end
 810
 811   defp restrict_muted_reblogs(query, _), do: query
 812
 813   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 814
 815   defp maybe_preload_objects(query, _) do
 816     query
 817     |> Activity.with_preloaded_object()
 818   end
 819
 820   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 821
 822   defp maybe_preload_bookmarks(query, opts) do
 823     query
 824     |> Activity.with_preloaded_bookmark(opts["user"])
 825   end
 826
 827   defp maybe_order(query, %{order: :desc}) do
 828     query
 829     |> order_by(desc: :id)
 830   end
 831
 832   defp maybe_order(query, %{order: :asc}) do
 833     query
 834     |> order_by(asc: :id)
 835   end
 836
 837   defp maybe_order(query, _), do: query
 838
 839   def fetch_activities_query(recipients, opts \\ %{}) do
 840     base_query = from(activity in Activity)
 841
 842     base_query
 843     |> maybe_preload_objects(opts)
 844     |> maybe_preload_bookmarks(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_favorited_by(opts)
 855     |> restrict_blocked(opts)
 856     |> restrict_muted(opts)
 857     |> restrict_media(opts)
 858     |> restrict_visibility(opts)
 859     |> restrict_thread_visibility(opts)
 860     |> restrict_replies(opts)
 861     |> restrict_reblogs(opts)
 862     |> restrict_pinned(opts)
 863     |> restrict_muted_reblogs(opts)
 864     |> Activity.restrict_deactivated_users()
 865   end
 866
 867   def fetch_activities(recipients, opts \\ %{}) do
 868     fetch_activities_query(recipients, opts)
 869     |> Pagination.fetch_paginated(opts)
 870     |> Enum.reverse()
 871   end
 872
 873   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 874     fetch_activities_query([], opts)
 875     |> restrict_to_cc(recipients_to, recipients_cc)
 876     |> Pagination.fetch_paginated(opts)
 877     |> Enum.reverse()
 878   end
 879
 880   def upload(file, opts \\ []) do
 881     with {:ok, data} <- Upload.store(file, opts) do
 882       obj_data =
 883         if opts[:actor] do
 884           Map.put(data, "actor", opts[:actor])
 885         else
 886           data
 887         end
 888
 889       Repo.insert(%Object{data: obj_data})
 890     end
 891   end
 892
 893   def user_data_from_user_object(data) do
 894     avatar =
 895       data["icon"]["url"] &&
 896         %{
 897           "type" => "Image",
 898           "url" => [%{"href" => data["icon"]["url"]}]
 899         }
 900
 901     banner =
 902       data["image"]["url"] &&
 903         %{
 904           "type" => "Image",
 905           "url" => [%{"href" => data["image"]["url"]}]
 906         }
 907
 908     locked = data["manuallyApprovesFollowers"] || false
 909     data = Transmogrifier.maybe_fix_user_object(data)
 910
 911     user_data = %{
 912       ap_id: data["id"],
 913       info: %{
 914         "ap_enabled" => true,
 915         "source_data" => data,
 916         "banner" => banner,
 917         "locked" => locked
 918       },
 919       avatar: avatar,
 920       name: data["name"],
 921       follower_address: data["followers"],
 922       bio: data["summary"]
 923     }
 924
 925     # nickname can be nil because of virtual actors
 926     user_data =
 927       if data["preferredUsername"] do
 928         Map.put(
 929           user_data,
 930           :nickname,
 931           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 932         )
 933       else
 934         Map.put(user_data, :nickname, nil)
 935       end
 936
 937     {:ok, user_data}
 938   end
 939
 940   def fetch_and_prepare_user_from_ap_id(ap_id) do
 941     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
 942       user_data_from_user_object(data)
 943     else
 944       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 945     end
 946   end
 947
 948   def make_user_from_ap_id(ap_id) do
 949     if _user = User.get_cached_by_ap_id(ap_id) do
 950       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 951     else
 952       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 953         User.insert_or_update_user(data)
 954       else
 955         e -> {:error, e}
 956       end
 957     end
 958   end
 959
 960   def make_user_from_nickname(nickname) do
 961     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 962       make_user_from_ap_id(ap_id)
 963     else
 964       _e -> {:error, "No AP id in WebFinger"}
 965     end
 966   end
 967
 968   # filter out broken threads
 969   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 970     entire_thread_visible_for_user?(activity, user)
 971   end
 972
 973   # do post-processing on a specific activity
 974   def contain_activity(%Activity{} = activity, %User{} = user) do
 975     contain_broken_threads(activity, user)
 976   end
 977
 978   def fetch_direct_messages_query do
 979     Activity
 980     |> restrict_type(%{"type" => "Create"})
 981     |> restrict_visibility(%{visibility: "direct"})
 982     |> order_by([activity], asc: activity.id)
 983   end
 984 end