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_recipients(query, [], _user), do: query
 657
 658   defp restrict_recipients(query, recipients, nil) do
 659     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 660   end
 661
 662   defp restrict_recipients(query, recipients, user) do
 663     from(
 664       activity in query,
 665       where: fragment("? && ?", ^recipients, activity.recipients),
 666       or_where: activity.actor == ^user.ap_id
 667     )
 668   end
 669
 670   defp restrict_local(query, %{"local_only" => true}) do
 671     from(activity in query, where: activity.local == true)
 672   end
 673
 674   defp restrict_local(query, _), do: query
 675
 676   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 677     from(activity in query, where: activity.actor == ^actor_id)
 678   end
 679
 680   defp restrict_actor(query, _), do: query
 681
 682   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 683     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 684   end
 685
 686   defp restrict_type(query, %{"type" => type}) do
 687     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 688   end
 689
 690   defp restrict_type(query, _), do: query
 691
 692   defp restrict_state(query, %{"state" => state}) do
 693     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 694   end
 695
 696   defp restrict_state(query, _), do: query
 697
 698   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 699     from(
 700       activity in query,
 701       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 702     )
 703   end
 704
 705   defp restrict_favorited_by(query, _), do: query
 706
 707   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 708     raise "Can't use the child object without preloading!"
 709   end
 710
 711   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 712     from(
 713       [_activity, object] in query,
 714       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 715     )
 716   end
 717
 718   defp restrict_media(query, _), do: query
 719
 720   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 721     from(
 722       activity in query,
 723       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 724     )
 725   end
 726
 727   defp restrict_replies(query, _), do: query
 728
 729   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 730     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 731   end
 732
 733   defp restrict_reblogs(query, _), do: query
 734
 735   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 736
 737   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 738     mutes = info.mutes
 739
 740     from(
 741       activity in query,
 742       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 743       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 744     )
 745   end
 746
 747   defp restrict_muted(query, _), do: query
 748
 749   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 750     blocks = info.blocks || []
 751     domain_blocks = info.domain_blocks || []
 752
 753     query =
 754       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 755
 756     from(
 757       [activity, object: o] in query,
 758       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 759       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 760       where:
 761         fragment(
 762           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 763           activity.data,
 764           activity.data,
 765           ^blocks
 766         ),
 767       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 768       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 769     )
 770   end
 771
 772   defp restrict_blocked(query, _), do: query
 773
 774   defp restrict_unlisted(query) do
 775     from(
 776       activity in query,
 777       where:
 778         fragment(
 779           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 780           activity.data,
 781           ^["https://www.w3.org/ns/activitystreams#Public"]
 782         )
 783     )
 784   end
 785
 786   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 787     from(activity in query, where: activity.id in ^ids)
 788   end
 789
 790   defp restrict_pinned(query, _), do: query
 791
 792   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 793     muted_reblogs = info.muted_reblogs || []
 794
 795     from(
 796       activity in query,
 797       where:
 798         fragment(
 799           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 800           activity.data,
 801           activity.actor,
 802           ^muted_reblogs
 803         )
 804     )
 805   end
 806
 807   defp restrict_muted_reblogs(query, _), do: query
 808
 809   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 810
 811   defp maybe_preload_objects(query, _) do
 812     query
 813     |> Activity.with_preloaded_object()
 814   end
 815
 816   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 817
 818   defp maybe_preload_bookmarks(query, opts) do
 819     query
 820     |> Activity.with_preloaded_bookmark(opts["user"])
 821   end
 822
 823   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 824
 825   defp maybe_set_thread_muted_field(query, opts) do
 826     query
 827     |> Activity.with_set_thread_muted_field(opts["user"])
 828   end
 829
 830   defp maybe_order(query, %{order: :desc}) do
 831     query
 832     |> order_by(desc: :id)
 833   end
 834
 835   defp maybe_order(query, %{order: :asc}) do
 836     query
 837     |> order_by(asc: :id)
 838   end
 839
 840   defp maybe_order(query, _), do: query
 841
 842   def fetch_activities_query(recipients, opts \\ %{}) do
 843     base_query = from(activity in Activity)
 844
 845     base_query
 846     |> maybe_preload_objects(opts)
 847     |> maybe_preload_bookmarks(opts)
 848     |> maybe_set_thread_muted_field(opts)
 849     |> maybe_order(opts)
 850     |> restrict_recipients(recipients, opts["user"])
 851     |> restrict_tag(opts)
 852     |> restrict_tag_reject(opts)
 853     |> restrict_tag_all(opts)
 854     |> restrict_since(opts)
 855     |> restrict_local(opts)
 856     |> restrict_actor(opts)
 857     |> restrict_type(opts)
 858     |> restrict_state(opts)
 859     |> restrict_favorited_by(opts)
 860     |> restrict_blocked(opts)
 861     |> restrict_muted(opts)
 862     |> restrict_media(opts)
 863     |> restrict_visibility(opts)
 864     |> restrict_thread_visibility(opts)
 865     |> restrict_replies(opts)
 866     |> restrict_reblogs(opts)
 867     |> restrict_pinned(opts)
 868     |> restrict_muted_reblogs(opts)
 869     |> Activity.restrict_deactivated_users()
 870   end
 871
 872   def fetch_activities(recipients, opts \\ %{}) do
 873     fetch_activities_query(recipients, opts)
 874     |> Pagination.fetch_paginated(opts)
 875     |> Enum.reverse()
 876   end
 877
 878   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
 879     from(activity in query,
 880       where:
 881         fragment("? && ?", activity.recipients, ^recipients) or
 882           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
 883              "https://www.w3.org/ns/activitystreams#Public" in activity.recipients)
 884     )
 885   end
 886
 887   def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
 888     fetch_activities_query([], opts)
 889     |> fetch_activities_bounded_query(recipients, recipients_with_public)
 890     |> Pagination.fetch_paginated(opts)
 891     |> Enum.reverse()
 892   end
 893
 894   def upload(file, opts \\ []) do
 895     with {:ok, data} <- Upload.store(file, opts) do
 896       obj_data =
 897         if opts[:actor] do
 898           Map.put(data, "actor", opts[:actor])
 899         else
 900           data
 901         end
 902
 903       Repo.insert(%Object{data: obj_data})
 904     end
 905   end
 906
 907   defp object_to_user_data(data) do
 908     avatar =
 909       data["icon"]["url"] &&
 910         %{
 911           "type" => "Image",
 912           "url" => [%{"href" => data["icon"]["url"]}]
 913         }
 914
 915     banner =
 916       data["image"]["url"] &&
 917         %{
 918           "type" => "Image",
 919           "url" => [%{"href" => data["image"]["url"]}]
 920         }
 921
 922     locked = data["manuallyApprovesFollowers"] || false
 923     data = Transmogrifier.maybe_fix_user_object(data)
 924
 925     user_data = %{
 926       ap_id: data["id"],
 927       info: %{
 928         "ap_enabled" => true,
 929         "source_data" => data,
 930         "banner" => banner,
 931         "locked" => locked
 932       },
 933       avatar: avatar,
 934       name: data["name"],
 935       follower_address: data["followers"],
 936       bio: data["summary"]
 937     }
 938
 939     # nickname can be nil because of virtual actors
 940     user_data =
 941       if data["preferredUsername"] do
 942         Map.put(
 943           user_data,
 944           :nickname,
 945           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 946         )
 947       else
 948         Map.put(user_data, :nickname, nil)
 949       end
 950
 951     {:ok, user_data}
 952   end
 953
 954   def user_data_from_user_object(data) do
 955     with {:ok, data} <- MRF.filter(data),
 956          {:ok, data} <- object_to_user_data(data) do
 957       {:ok, data}
 958     else
 959       e -> {:error, e}
 960     end
 961   end
 962
 963   def fetch_and_prepare_user_from_ap_id(ap_id) do
 964     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
 965          {:ok, data} <- user_data_from_user_object(data) do
 966       {:ok, data}
 967     else
 968       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 969     end
 970   end
 971
 972   def make_user_from_ap_id(ap_id) do
 973     if _user = User.get_cached_by_ap_id(ap_id) do
 974       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 975     else
 976       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 977         User.insert_or_update_user(data)
 978       else
 979         e -> {:error, e}
 980       end
 981     end
 982   end
 983
 984   def make_user_from_nickname(nickname) do
 985     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 986       make_user_from_ap_id(ap_id)
 987     else
 988       _e -> {:error, "No AP id in WebFinger"}
 989     end
 990   end
 991
 992   # filter out broken threads
 993   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 994     entire_thread_visible_for_user?(activity, user)
 995   end
 996
 997   # do post-processing on a specific activity
 998   def contain_activity(%Activity{} = activity, %User{} = user) do
 999     contain_broken_threads(activity, user)
1000   end
1001
1002   def fetch_direct_messages_query do
1003     Activity
1004     |> restrict_type(%{"type" => "Create"})
1005     |> restrict_visibility(%{visibility: "direct"})
1006     |> order_by([activity], asc: activity.id)
1007   end
1008 end