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.Instances
   8   alias Pleroma.Notification
   9   alias Pleroma.Object
  10   alias Pleroma.Repo
  11   alias Pleroma.Upload
  12   alias Pleroma.User
  13   alias Pleroma.Web.ActivityPub.MRF
  14   alias Pleroma.Web.ActivityPub.Transmogrifier
  15   alias Pleroma.Web.Federator
  16   alias Pleroma.Web.OStatus
  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   @httpoison Application.get_env(:pleroma, :httpoison)
  26
  27   # For Announce activities, we filter the recipients based on following status for any actors
  28   # that match actual users.  See issue #164 for more information about why this is necessary.
  29   defp get_recipients(%{"type" => "Announce"} = data) do
  30     to = data["to"] || []
  31     cc = data["cc"] || []
  32     actor = User.get_cached_by_ap_id(data["actor"])
  33
  34     recipients =
  35       (to ++ cc)
  36       |> Enum.filter(fn recipient ->
  37         case User.get_cached_by_ap_id(recipient) do
  38           nil ->
  39             true
  40
  41           user ->
  42             User.following?(user, actor)
  43         end
  44       end)
  45
  46     {recipients, to, cc}
  47   end
  48
  49   defp get_recipients(%{"type" => "Create"} = data) do
  50     to = data["to"] || []
  51     cc = data["cc"] || []
  52     actor = data["actor"] || []
  53     recipients = (to ++ cc ++ [actor]) |> Enum.uniq()
  54     {recipients, to, cc}
  55   end
  56
  57   defp get_recipients(data) do
  58     to = data["to"] || []
  59     cc = data["cc"] || []
  60     recipients = to ++ cc
  61     {recipients, to, cc}
  62   end
  63
  64   defp check_actor_is_active(actor) do
  65     if not is_nil(actor) do
  66       with user <- User.get_cached_by_ap_id(actor),
  67            false <- user.info.deactivated do
  68         :ok
  69       else
  70         _e -> :reject
  71       end
  72     else
  73       :ok
  74     end
  75   end
  76
  77   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
  78     limit = Pleroma.Config.get([:instance, :remote_limit])
  79     String.length(content) <= limit
  80   end
  81
  82   defp check_remote_limit(_), do: true
  83
  84   def increase_note_count_if_public(actor, object) do
  85     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  86   end
  87
  88   def decrease_note_count_if_public(actor, object) do
  89     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  90   end
  91
  92   def insert(map, local \\ true) when is_map(map) do
  93     with nil <- Activity.normalize(map),
  94          map <- lazy_put_activity_defaults(map),
  95          :ok <- check_actor_is_active(map["actor"]),
  96          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
  97          {:ok, map} <- MRF.filter(map),
  98          :ok <- insert_full_object(map) do
  99       {recipients, _, _} = get_recipients(map)
 100
 101       {:ok, activity} =
 102         Repo.insert(%Activity{
 103           data: map,
 104           local: local,
 105           actor: map["actor"],
 106           recipients: recipients
 107         })
 108
 109       Task.start(fn ->
 110         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 111       end)
 112
 113       Notification.create_notifications(activity)
 114       stream_out(activity)
 115       {:ok, activity}
 116     else
 117       %Activity{} = activity -> {:ok, activity}
 118       error -> {:error, error}
 119     end
 120   end
 121
 122   def stream_out(activity) do
 123     public = "https://www.w3.org/ns/activitystreams#Public"
 124
 125     if activity.data["type"] in ["Create", "Announce", "Delete"] do
 126       Pleroma.Web.Streamer.stream("user", activity)
 127       Pleroma.Web.Streamer.stream("list", activity)
 128
 129       if Enum.member?(activity.data["to"], public) do
 130         Pleroma.Web.Streamer.stream("public", activity)
 131
 132         if activity.local do
 133           Pleroma.Web.Streamer.stream("public:local", activity)
 134         end
 135
 136         if activity.data["type"] in ["Create"] do
 137           activity.data["object"]
 138           |> Map.get("tag", [])
 139           |> Enum.filter(fn tag -> is_bitstring(tag) end)
 140           |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
 141
 142           if activity.data["object"]["attachment"] != [] do
 143             Pleroma.Web.Streamer.stream("public:media", activity)
 144
 145             if activity.local do
 146               Pleroma.Web.Streamer.stream("public:local:media", activity)
 147             end
 148           end
 149         end
 150       else
 151         if !Enum.member?(activity.data["cc"] || [], public) &&
 152              !Enum.member?(
 153                activity.data["to"],
 154                User.get_by_ap_id(activity.data["actor"]).follower_address
 155              ),
 156            do: Pleroma.Web.Streamer.stream("direct", activity)
 157       end
 158     end
 159   end
 160
 161   def create(%{to: to, actor: actor, context: context, object: object} = params) do
 162     additional = params[:additional] || %{}
 163     # only accept false as false value
 164     local = !(params[:local] == false)
 165     published = params[:published]
 166
 167     with create_data <-
 168            make_create_data(
 169              %{to: to, actor: actor, published: published, context: context, object: object},
 170              additional
 171            ),
 172          {:ok, activity} <- insert(create_data, local),
 173          # Changing note count prior to enqueuing federation task in order to avoid
 174          # race conditions on updating user.info
 175          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 176          :ok <- maybe_federate(activity) do
 177       {:ok, activity}
 178     end
 179   end
 180
 181   def accept(%{to: to, actor: actor, object: object} = params) do
 182     # only accept false as false value
 183     local = !(params[:local] == false)
 184
 185     with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
 186          {:ok, activity} <- insert(data, local),
 187          :ok <- maybe_federate(activity) do
 188       {:ok, activity}
 189     end
 190   end
 191
 192   def reject(%{to: to, actor: actor, object: object} = params) do
 193     # only accept false as false value
 194     local = !(params[:local] == false)
 195
 196     with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
 197          {:ok, activity} <- insert(data, local),
 198          :ok <- maybe_federate(activity) do
 199       {:ok, activity}
 200     end
 201   end
 202
 203   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 204     # only accept false as false value
 205     local = !(params[:local] == false)
 206
 207     with data <- %{
 208            "to" => to,
 209            "cc" => cc,
 210            "type" => "Update",
 211            "actor" => actor,
 212            "object" => object
 213          },
 214          {:ok, activity} <- insert(data, local),
 215          :ok <- maybe_federate(activity) do
 216       {:ok, activity}
 217     end
 218   end
 219
 220   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 221   def like(
 222         %User{ap_id: ap_id} = user,
 223         %Object{data: %{"id" => _}} = object,
 224         activity_id \\ nil,
 225         local \\ true
 226       ) do
 227     with nil <- get_existing_like(ap_id, object),
 228          like_data <- make_like_data(user, object, activity_id),
 229          {:ok, activity} <- insert(like_data, local),
 230          {:ok, object} <- add_like_to_object(activity, object),
 231          :ok <- maybe_federate(activity) do
 232       {:ok, activity, object}
 233     else
 234       %Activity{} = activity -> {:ok, activity, object}
 235       error -> {:error, error}
 236     end
 237   end
 238
 239   def unlike(
 240         %User{} = actor,
 241         %Object{} = object,
 242         activity_id \\ nil,
 243         local \\ true
 244       ) do
 245     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 246          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 247          {:ok, unlike_activity} <- insert(unlike_data, local),
 248          {:ok, _activity} <- Repo.delete(like_activity),
 249          {:ok, object} <- remove_like_from_object(like_activity, object),
 250          :ok <- maybe_federate(unlike_activity) do
 251       {:ok, unlike_activity, like_activity, object}
 252     else
 253       _e -> {:ok, object}
 254     end
 255   end
 256
 257   def announce(
 258         %User{ap_id: _} = user,
 259         %Object{data: %{"id" => _}} = object,
 260         activity_id \\ nil,
 261         local \\ true,
 262         public \\ true
 263       ) do
 264     with true <- is_public?(object),
 265          announce_data <- make_announce_data(user, object, activity_id, public),
 266          {:ok, activity} <- insert(announce_data, local),
 267          {:ok, object} <- add_announce_to_object(activity, object),
 268          :ok <- maybe_federate(activity) do
 269       {:ok, activity, object}
 270     else
 271       error -> {:error, error}
 272     end
 273   end
 274
 275   def unannounce(
 276         %User{} = actor,
 277         %Object{} = object,
 278         activity_id \\ nil,
 279         local \\ true
 280       ) do
 281     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 282          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 283          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 284          :ok <- maybe_federate(unannounce_activity),
 285          {:ok, _activity} <- Repo.delete(announce_activity),
 286          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 287       {:ok, unannounce_activity, object}
 288     else
 289       _e -> {:ok, object}
 290     end
 291   end
 292
 293   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 294     with data <- make_follow_data(follower, followed, activity_id),
 295          {:ok, activity} <- insert(data, local),
 296          :ok <- maybe_federate(activity) do
 297       {:ok, activity}
 298     end
 299   end
 300
 301   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 302     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 303          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 304          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 305          {:ok, activity} <- insert(unfollow_data, local),
 306          :ok <- maybe_federate(activity) do
 307       {:ok, activity}
 308     end
 309   end
 310
 311   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
 312     user = User.get_cached_by_ap_id(actor)
 313     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 314
 315     with {:ok, object, activity} <- Object.delete(object),
 316          data <- %{
 317            "type" => "Delete",
 318            "actor" => actor,
 319            "object" => id,
 320            "to" => to,
 321            "deleted_activity_id" => activity && activity.id
 322          },
 323          {:ok, activity} <- insert(data, local),
 324          # Changing note count prior to enqueuing federation task in order to avoid
 325          # race conditions on updating user.info
 326          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 327          :ok <- maybe_federate(activity) do
 328       {:ok, activity}
 329     end
 330   end
 331
 332   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 333     ap_config = Application.get_env(:pleroma, :activitypub)
 334     unfollow_blocked = Keyword.get(ap_config, :unfollow_blocked)
 335     outgoing_blocks = Keyword.get(ap_config, :outgoing_blocks)
 336
 337     with true <- unfollow_blocked do
 338       follow_activity = fetch_latest_follow(blocker, blocked)
 339
 340       if follow_activity do
 341         unfollow(blocker, blocked, nil, local)
 342       end
 343     end
 344
 345     with true <- outgoing_blocks,
 346          block_data <- make_block_data(blocker, blocked, activity_id),
 347          {:ok, activity} <- insert(block_data, local),
 348          :ok <- maybe_federate(activity) do
 349       {:ok, activity}
 350     else
 351       _e -> {:ok, nil}
 352     end
 353   end
 354
 355   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 356     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 357          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 358          {:ok, activity} <- insert(unblock_data, local),
 359          :ok <- maybe_federate(activity) do
 360       {:ok, activity}
 361     end
 362   end
 363
 364   def flag(
 365         %{
 366           actor: actor,
 367           context: context,
 368           account: account,
 369           statuses: statuses,
 370           content: content
 371         } = params
 372       ) do
 373     # only accept false as false value
 374     local = !(params[:local] == false)
 375     forward = !(params[:forward] == false)
 376
 377     additional = params[:additional] || %{}
 378
 379     params = %{
 380       actor: actor,
 381       context: context,
 382       account: account,
 383       statuses: statuses,
 384       content: content
 385     }
 386
 387     additional =
 388       if forward do
 389         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 390       else
 391         Map.merge(additional, %{"to" => [], "cc" => []})
 392       end
 393
 394     with flag_data <- make_flag_data(params, additional),
 395          {:ok, activity} <- insert(flag_data, local),
 396          :ok <- maybe_federate(activity) do
 397       Enum.each(User.all_superusers(), fn superuser ->
 398         superuser
 399         |> Pleroma.AdminEmail.report(actor, account, statuses, content)
 400         |> Pleroma.Mailer.deliver_async()
 401       end)
 402
 403       {:ok, activity}
 404     end
 405   end
 406
 407   def fetch_activities_for_context(context, opts \\ %{}) do
 408     public = ["https://www.w3.org/ns/activitystreams#Public"]
 409
 410     recipients =
 411       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 412
 413     query = from(activity in Activity)
 414
 415     query =
 416       query
 417       |> restrict_blocked(opts)
 418       |> restrict_recipients(recipients, opts["user"])
 419
 420     query =
 421       from(
 422         activity in query,
 423         where:
 424           fragment(
 425             "?->>'type' = ? and ?->>'context' = ?",
 426             activity.data,
 427             "Create",
 428             activity.data,
 429             ^context
 430           ),
 431         order_by: [desc: :id]
 432       )
 433
 434     Repo.all(query)
 435   end
 436
 437   def fetch_public_activities(opts \\ %{}) do
 438     q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
 439
 440     q
 441     |> restrict_unlisted()
 442     |> Repo.all()
 443     |> Enum.reverse()
 444   end
 445
 446   @valid_visibilities ~w[direct unlisted public private]
 447
 448   defp restrict_visibility(query, %{visibility: visibility})
 449        when is_list(visibility) do
 450     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 451       query =
 452         from(
 453           a in query,
 454           where:
 455             fragment(
 456               "activity_visibility(?, ?, ?) = ANY (?)",
 457               a.actor,
 458               a.recipients,
 459               a.data,
 460               ^visibility
 461             )
 462         )
 463
 464       Ecto.Adapters.SQL.to_sql(:all, Repo, query)
 465
 466       query
 467     else
 468       Logger.error("Could not restrict visibility to #{visibility}")
 469     end
 470   end
 471
 472   defp restrict_visibility(query, %{visibility: visibility})
 473        when visibility in @valid_visibilities do
 474     query =
 475       from(
 476         a in query,
 477         where:
 478           fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 479       )
 480
 481     Ecto.Adapters.SQL.to_sql(:all, Repo, query)
 482
 483     query
 484   end
 485
 486   defp restrict_visibility(_query, %{visibility: visibility})
 487        when visibility not in @valid_visibilities do
 488     Logger.error("Could not restrict visibility to #{visibility}")
 489   end
 490
 491   defp restrict_visibility(query, _visibility), do: query
 492
 493   def fetch_user_activities(user, reading_user, params \\ %{}) do
 494     params =
 495       params
 496       |> Map.put("type", ["Create", "Announce"])
 497       |> Map.put("actor_id", user.ap_id)
 498       |> Map.put("whole_db", true)
 499       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 500
 501     recipients =
 502       if reading_user do
 503         ["https://www.w3.org/ns/activitystreams#Public"] ++
 504           [reading_user.ap_id | reading_user.following]
 505       else
 506         ["https://www.w3.org/ns/activitystreams#Public"]
 507       end
 508
 509     fetch_activities(recipients, params)
 510     |> Enum.reverse()
 511   end
 512
 513   defp restrict_since(query, %{"since_id" => ""}), do: query
 514
 515   defp restrict_since(query, %{"since_id" => since_id}) do
 516     from(activity in query, where: activity.id > ^since_id)
 517   end
 518
 519   defp restrict_since(query, _), do: query
 520
 521   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 522        when is_list(tag_reject) and tag_reject != [] do
 523     from(
 524       activity in query,
 525       where: fragment(~s(\(not \(? #> '{"object","tag"}'\) \\?| ?\)), activity.data, ^tag_reject)
 526     )
 527   end
 528
 529   defp restrict_tag_reject(query, _), do: query
 530
 531   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 532        when is_list(tag_all) and tag_all != [] do
 533     from(
 534       activity in query,
 535       where: fragment(~s(\(? #> '{"object","tag"}'\) \\?& ?), activity.data, ^tag_all)
 536     )
 537   end
 538
 539   defp restrict_tag_all(query, _), do: query
 540
 541   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 542     from(
 543       activity in query,
 544       where: fragment(~s(\(? #> '{"object","tag"}'\) \\?| ?), activity.data, ^tag)
 545     )
 546   end
 547
 548   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 549     from(
 550       activity in query,
 551       where: fragment(~s(? <@ (? #> '{"object","tag"}'\)), ^tag, activity.data)
 552     )
 553   end
 554
 555   defp restrict_tag(query, _), do: query
 556
 557   defp restrict_to_cc(query, recipients_to, recipients_cc) do
 558     from(
 559       activity in query,
 560       where:
 561         fragment(
 562           "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
 563           activity.data,
 564           ^recipients_to,
 565           activity.data,
 566           ^recipients_cc
 567         )
 568     )
 569   end
 570
 571   defp restrict_recipients(query, [], _user), do: query
 572
 573   defp restrict_recipients(query, recipients, nil) do
 574     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 575   end
 576
 577   defp restrict_recipients(query, recipients, user) do
 578     from(
 579       activity in query,
 580       where: fragment("? && ?", ^recipients, activity.recipients),
 581       or_where: activity.actor == ^user.ap_id
 582     )
 583   end
 584
 585   defp restrict_limit(query, %{"limit" => limit}) do
 586     from(activity in query, limit: ^limit)
 587   end
 588
 589   defp restrict_limit(query, _), do: query
 590
 591   defp restrict_local(query, %{"local_only" => true}) do
 592     from(activity in query, where: activity.local == true)
 593   end
 594
 595   defp restrict_local(query, _), do: query
 596
 597   defp restrict_max(query, %{"max_id" => ""}), do: query
 598
 599   defp restrict_max(query, %{"max_id" => max_id}) do
 600     from(activity in query, where: activity.id < ^max_id)
 601   end
 602
 603   defp restrict_max(query, _), do: query
 604
 605   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 606     from(activity in query, where: activity.actor == ^actor_id)
 607   end
 608
 609   defp restrict_actor(query, _), do: query
 610
 611   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 612     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 613   end
 614
 615   defp restrict_type(query, %{"type" => type}) do
 616     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 617   end
 618
 619   defp restrict_type(query, _), do: query
 620
 621   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 622     from(
 623       activity in query,
 624       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 625     )
 626   end
 627
 628   defp restrict_favorited_by(query, _), do: query
 629
 630   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 631     from(
 632       activity in query,
 633       where: fragment(~s(not (? #> '{"object","attachment"}' = ?\)), activity.data, ^[])
 634     )
 635   end
 636
 637   defp restrict_media(query, _), do: query
 638
 639   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 640     from(
 641       activity in query,
 642       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 643     )
 644   end
 645
 646   defp restrict_replies(query, _), do: query
 647
 648   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 649     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 650   end
 651
 652   defp restrict_reblogs(query, _), do: query
 653
 654   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 655
 656   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 657     mutes = info.mutes
 658
 659     from(
 660       activity in query,
 661       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 662       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 663     )
 664   end
 665
 666   defp restrict_muted(query, _), do: query
 667
 668   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 669     blocks = info.blocks || []
 670     domain_blocks = info.domain_blocks || []
 671
 672     from(
 673       activity in query,
 674       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 675       where: fragment("not (?->'to' \\?| ?)", activity.data, ^blocks),
 676       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
 677     )
 678   end
 679
 680   defp restrict_blocked(query, _), do: query
 681
 682   defp restrict_unlisted(query) do
 683     from(
 684       activity in query,
 685       where:
 686         fragment(
 687           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 688           activity.data,
 689           ^["https://www.w3.org/ns/activitystreams#Public"]
 690         )
 691     )
 692   end
 693
 694   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 695     from(activity in query, where: activity.id in ^ids)
 696   end
 697
 698   defp restrict_pinned(query, _), do: query
 699
 700   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 701     muted_reblogs = info.muted_reblogs || []
 702
 703     from(
 704       activity in query,
 705       where: fragment("not ?->>'type' = 'Announce'", activity.data),
 706       where: fragment("not ? = ANY(?)", activity.actor, ^muted_reblogs)
 707     )
 708   end
 709
 710   defp restrict_muted_reblogs(query, _), do: query
 711
 712   def fetch_activities_query(recipients, opts \\ %{}) do
 713     base_query =
 714       from(
 715         activity in Activity,
 716         limit: 20,
 717         order_by: [fragment("? desc nulls last", activity.id)]
 718       )
 719
 720     base_query
 721     |> restrict_recipients(recipients, opts["user"])
 722     |> restrict_tag(opts)
 723     |> restrict_tag_reject(opts)
 724     |> restrict_tag_all(opts)
 725     |> restrict_since(opts)
 726     |> restrict_local(opts)
 727     |> restrict_limit(opts)
 728     |> restrict_max(opts)
 729     |> restrict_actor(opts)
 730     |> restrict_type(opts)
 731     |> restrict_favorited_by(opts)
 732     |> restrict_blocked(opts)
 733     |> restrict_muted(opts)
 734     |> restrict_media(opts)
 735     |> restrict_visibility(opts)
 736     |> restrict_replies(opts)
 737     |> restrict_reblogs(opts)
 738     |> restrict_pinned(opts)
 739     |> restrict_muted_reblogs(opts)
 740   end
 741
 742   def fetch_activities(recipients, opts \\ %{}) do
 743     fetch_activities_query(recipients, opts)
 744     |> Repo.all()
 745     |> Enum.reverse()
 746   end
 747
 748   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 749     fetch_activities_query([], opts)
 750     |> restrict_to_cc(recipients_to, recipients_cc)
 751     |> Repo.all()
 752     |> Enum.reverse()
 753   end
 754
 755   def upload(file, opts \\ []) do
 756     with {:ok, data} <- Upload.store(file, opts) do
 757       obj_data =
 758         if opts[:actor] do
 759           Map.put(data, "actor", opts[:actor])
 760         else
 761           data
 762         end
 763
 764       Repo.insert(%Object{data: obj_data})
 765     end
 766   end
 767
 768   def user_data_from_user_object(data) do
 769     avatar =
 770       data["icon"]["url"] &&
 771         %{
 772           "type" => "Image",
 773           "url" => [%{"href" => data["icon"]["url"]}]
 774         }
 775
 776     banner =
 777       data["image"]["url"] &&
 778         %{
 779           "type" => "Image",
 780           "url" => [%{"href" => data["image"]["url"]}]
 781         }
 782
 783     locked = data["manuallyApprovesFollowers"] || false
 784     data = Transmogrifier.maybe_fix_user_object(data)
 785
 786     user_data = %{
 787       ap_id: data["id"],
 788       info: %{
 789         "ap_enabled" => true,
 790         "source_data" => data,
 791         "banner" => banner,
 792         "locked" => locked
 793       },
 794       avatar: avatar,
 795       name: data["name"],
 796       follower_address: data["followers"],
 797       bio: data["summary"]
 798     }
 799
 800     # nickname can be nil because of virtual actors
 801     user_data =
 802       if data["preferredUsername"] do
 803         Map.put(
 804           user_data,
 805           :nickname,
 806           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 807         )
 808       else
 809         Map.put(user_data, :nickname, nil)
 810       end
 811
 812     {:ok, user_data}
 813   end
 814
 815   def fetch_and_prepare_user_from_ap_id(ap_id) do
 816     with {:ok, data} <- fetch_and_contain_remote_object_from_id(ap_id) do
 817       user_data_from_user_object(data)
 818     else
 819       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 820     end
 821   end
 822
 823   def make_user_from_ap_id(ap_id) do
 824     if _user = User.get_by_ap_id(ap_id) do
 825       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 826     else
 827       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 828         User.insert_or_update_user(data)
 829       else
 830         e -> {:error, e}
 831       end
 832     end
 833   end
 834
 835   def make_user_from_nickname(nickname) do
 836     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 837       make_user_from_ap_id(ap_id)
 838     else
 839       _e -> {:error, "No AP id in WebFinger"}
 840     end
 841   end
 842
 843   def should_federate?(inbox, public) do
 844     if public do
 845       true
 846     else
 847       inbox_info = URI.parse(inbox)
 848       !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host)
 849     end
 850   end
 851
 852   def publish(actor, activity) do
 853     remote_followers =
 854       if actor.follower_address in activity.recipients do
 855         {:ok, followers} = User.get_followers(actor)
 856         followers |> Enum.filter(&(!&1.local))
 857       else
 858         []
 859       end
 860
 861     public = is_public?(activity)
 862
 863     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 864     json = Jason.encode!(data)
 865
 866     (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers)
 867     |> Enum.filter(fn user -> User.ap_enabled?(user) end)
 868     |> Enum.map(fn %{info: %{source_data: data}} ->
 869       (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
 870     end)
 871     |> Enum.uniq()
 872     |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
 873     |> Instances.filter_reachable()
 874     |> Enum.each(fn {inbox, unreachable_since} ->
 875       Federator.publish_single_ap(%{
 876         inbox: inbox,
 877         json: json,
 878         actor: actor,
 879         id: activity.data["id"],
 880         unreachable_since: unreachable_since
 881       })
 882     end)
 883   end
 884
 885   def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do
 886     Logger.info("Federating #{id} to #{inbox}")
 887     host = URI.parse(inbox).host
 888
 889     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
 890
 891     date =
 892       NaiveDateTime.utc_now()
 893       |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
 894
 895     signature =
 896       Pleroma.Web.HTTPSignatures.sign(actor, %{
 897         host: host,
 898         "content-length": byte_size(json),
 899         digest: digest,
 900         date: date
 901       })
 902
 903     with {:ok, %{status: code}} when code in 200..299 <-
 904            result =
 905              @httpoison.post(
 906                inbox,
 907                json,
 908                [
 909                  {"Content-Type", "application/activity+json"},
 910                  {"Date", date},
 911                  {"signature", signature},
 912                  {"digest", digest}
 913                ]
 914              ) do
 915       if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
 916         do: Instances.set_reachable(inbox)
 917
 918       result
 919     else
 920       {_post_result, response} ->
 921         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
 922         {:error, response}
 923     end
 924   end
 925
 926   # TODO:
 927   # This will create a Create activity, which we need internally at the moment.
 928   def fetch_object_from_id(id) do
 929     if object = Object.get_cached_by_ap_id(id) do
 930       {:ok, object}
 931     else
 932       with {:ok, data} <- fetch_and_contain_remote_object_from_id(id),
 933            nil <- Object.normalize(data),
 934            params <- %{
 935              "type" => "Create",
 936              "to" => data["to"],
 937              "cc" => data["cc"],
 938              "actor" => data["actor"] || data["attributedTo"],
 939              "object" => data
 940            },
 941            :ok <- Transmogrifier.contain_origin(id, params),
 942            {:ok, activity} <- Transmogrifier.handle_incoming(params) do
 943         {:ok, Object.normalize(activity.data["object"])}
 944       else
 945         {:error, {:reject, nil}} ->
 946           {:reject, nil}
 947
 948         object = %Object{} ->
 949           {:ok, object}
 950
 951         _e ->
 952           Logger.info("Couldn't get object via AP, trying out OStatus fetching...")
 953
 954           case OStatus.fetch_activity_from_url(id) do
 955             {:ok, [activity | _]} -> {:ok, Object.normalize(activity.data["object"])}
 956             e -> e
 957           end
 958       end
 959     end
 960   end
 961
 962   def fetch_and_contain_remote_object_from_id(id) do
 963     Logger.info("Fetching object #{id} via AP")
 964
 965     with true <- String.starts_with?(id, "http"),
 966          {:ok, %{body: body, status: code}} when code in 200..299 <-
 967            @httpoison.get(
 968              id,
 969              [{:Accept, "application/activity+json"}]
 970            ),
 971          {:ok, data} <- Jason.decode(body),
 972          :ok <- Transmogrifier.contain_origin_from_id(id, data) do
 973       {:ok, data}
 974     else
 975       e ->
 976         {:error, e}
 977     end
 978   end
 979
 980   # filter out broken threads
 981   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 982     entire_thread_visible_for_user?(activity, user)
 983   end
 984
 985   # do post-processing on a specific activity
 986   def contain_activity(%Activity{} = activity, %User{} = user) do
 987     contain_broken_threads(activity, user)
 988   end
 989
 990   # do post-processing on a timeline
 991   def contain_timeline(timeline, user) do
 992     timeline
 993     |> Enum.filter(fn activity ->
 994       contain_activity(activity, user)
 995     end)
 996   end
 997 end