Merge branch 'develop' into issue/733

25 files changed, 887 insertions, 502 deletions

diff --git a/lib/pleroma/activity/ir/topics.ex b/lib/pleroma/activity/ir/topics.ex
new file mode 100644
index 000000000..010897abc
--- /dev/null
+++ b/lib/pleroma/activity/ir/topics.ex
@@ -0,0 +1,63 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Activity.Ir.Topics do
+  alias Pleroma.Object
+  alias Pleroma.Web.ActivityPub.Visibility
+
+  def get_activity_topics(activity) do
+    activity
+    |> Object.normalize()
+    |> generate_topics(activity)
+    |> List.flatten()
+  end
+
+  defp generate_topics(%{data: %{"type" => "Answer"}}, _) do
+    []
+  end
+
+  defp generate_topics(object, activity) do
+    ["user", "list"] ++ visibility_tags(object, activity)
+  end
+
+  defp visibility_tags(object, activity) do
+    case Visibility.get_visibility(activity) do
+      "public" ->
+        if activity.local do
+          ["public", "public:local"]
+        else
+          ["public"]
+        end
+        |> item_creation_tags(object, activity)
+
+      "direct" ->
+        ["direct"]
+
+      _ ->
+        []
+    end
+  end
+
+  defp item_creation_tags(tags, %{data: %{"type" => "Create"}} = object, activity) do
+    tags ++ hashtags_to_topics(object) ++ attachment_topics(object, activity)
+  end
+
+  defp item_creation_tags(tags, _, _) do
+    tags
+  end
+
+  defp hashtags_to_topics(%{data: %{"tag" => tags}}) do
+    tags
+    |> Enum.filter(&is_bitstring(&1))
+    |> Enum.map(fn tag -> "hashtag:" <> tag end)
+  end
+
+  defp hashtags_to_topics(_), do: []
+
+  defp attachment_topics(%{data: %{"attachment" => []}}, _act), do: []
+
+  defp attachment_topics(_object, %{local: true}), do: ["public:media", "public:local:media"]
+
+  defp attachment_topics(_object, _act), do: ["public:media"]
+end
diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 49094704b..dabce771d 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -43,23 +43,9 @@ defmodule Pleroma.Application do
         hackney_pool_children() ++
         [
           Pleroma.Stats,
-          {Oban, Pleroma.Config.get(Oban)},
-          %{
-            id: :web_push_init,
-            start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
-            restart: :temporary
-          },
-          %{
-            id: :federator_init,
-            start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]},
-            restart: :temporary
-          },
-          %{
-            id: :internal_fetch_init,
-            start: {Task, :start_link, [&Pleroma.Web.ActivityPub.InternalFetchActor.init/0]},
-            restart: :temporary
-          }
+          {Oban, Pleroma.Config.get(Oban)}
         ] ++
+        task_children(@env) ++
         oauth_cleanup_child(oauth_cleanup_enabled?()) ++
         streamer_child(@env) ++
         chat_child(@env, chat_enabled?()) ++
@@ -141,7 +127,7 @@ defmodule Pleroma.Application do
   defp streamer_child(:test), do: []
 
   defp streamer_child(_) do
-    [Pleroma.Web.Streamer]
+    [Pleroma.Web.Streamer.supervisor()]
   end
 
   defp oauth_cleanup_child(true),
@@ -163,4 +149,39 @@ defmodule Pleroma.Application do
       :hackney_pool.child_spec(pool, options)
     end
   end
+
+  defp task_children(:test) do
+    [
+      %{
+        id: :web_push_init,
+        start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
+        restart: :temporary
+      },
+      %{
+        id: :federator_init,
+        start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]},
+        restart: :temporary
+      }
+    ]
+  end
+
+  defp task_children(_) do
+    [
+      %{
+        id: :web_push_init,
+        start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
+        restart: :temporary
+      },
+      %{
+        id: :federator_init,
+        start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]},
+        restart: :temporary
+      },
+      %{
+        id: :internal_fetch_init,
+        start: {Task, :start_link, [&Pleroma.Web.ActivityPub.InternalFetchActor.init/0]},
+        restart: :temporary
+      }
+    ]
+  end
 end
diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex
new file mode 100644
index 000000000..29a1e5a77
--- /dev/null
+++ b/lib/pleroma/delivery.ex
@@ -0,0 +1,51 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Delivery do
+  use Ecto.Schema
+
+  alias Pleroma.Delivery
+  alias Pleroma.FlakeId
+  alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.User
+  alias Pleroma.User
+
+  import Ecto.Changeset
+  import Ecto.Query
+
+  schema "deliveries" do
+    belongs_to(:user, User, type: FlakeId)
+    belongs_to(:object, Object)
+  end
+
+  def changeset(delivery, params \\ %{}) do
+    delivery
+    |> cast(params, [:user_id, :object_id])
+    |> validate_required([:user_id, :object_id])
+    |> foreign_key_constraint(:object_id)
+    |> foreign_key_constraint(:user_id)
+    |> unique_constraint(:user_id, name: :deliveries_user_id_object_id_index)
+  end
+
+  def create(object_id, user_id) do
+    %Delivery{}
+    |> changeset(%{user_id: user_id, object_id: object_id})
+    |> Repo.insert(on_conflict: :nothing)
+  end
+
+  def get(object_id, user_id) do
+    from(d in Delivery, where: d.user_id == ^user_id and d.object_id == ^object_id)
+    |> Repo.one()
+  end
+
+  # A hack because user delete activities have a fake id for whatever reason
+  # TODO: Get rid of this
+  def delete_all_by_object_id("pleroma:fake_object_id"), do: {0, []}
+
+  def delete_all_by_object_id(object_id) do
+    from(d in Delivery, where: d.object_id == ^object_id)
+    |> Repo.delete_all()
+  end
+end
diff --git a/lib/pleroma/flake_id.ex b/lib/pleroma/flake_id.ex
index 47d61ca5f..042cf8659 100644
--- a/lib/pleroma/flake_id.ex
+++ b/lib/pleroma/flake_id.ex
@@ -14,7 +14,7 @@ defmodule Pleroma.FlakeId do
 
   @type t :: binary
 
-  @behaviour Ecto.Type
+  use Ecto.Type
   use GenServer
   require Logger
   alias __MODULE__
diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex
index b7c880c51..8012389ac 100644
--- a/lib/pleroma/notification.ex
+++ b/lib/pleroma/notification.ex
@@ -210,8 +210,10 @@ defmodule Pleroma.Notification do
     unless skip?(activity, user) do
       notification = %Notification{user_id: user.id, activity: activity}
       {:ok, notification} = Repo.insert(notification)
-      Streamer.stream("user", notification)
-      Streamer.stream("user:notification", notification)
+
+      ["user", "user:notification"]
+      |> Streamer.stream(notification)
+
       Push.send(notification)
       notification
     end
diff --git a/lib/pleroma/plugs/cache.ex b/lib/pleroma/plugs/cache.ex
index a81a861d0..50b534e7b 100644
--- a/lib/pleroma/plugs/cache.ex
+++ b/lib/pleroma/plugs/cache.ex
@@ -20,6 +20,7 @@ defmodule Pleroma.Plugs.Cache do
 
   - `ttl`:  An expiration time (time-to-live). This value should be in milliseconds or `nil` to disable expiration. Defaults to `nil`.
   - `query_params`: Take URL query string into account (`true`), ignore it (`false`) or limit to specific params only (list). Defaults to `true`.
+  - `tracking_fun`: A function that is called on successfull responses, no matter if the request is cached or not. It should accept a conn as the first argument and the value assigned to `tracking_fun_data` as the second.
 
   Additionally, you can overwrite the TTL inside a controller action by assigning `cache_ttl` to the connection struct:
 
@@ -56,6 +57,11 @@ defmodule Pleroma.Plugs.Cache do
       {:ok, nil} ->
         cache_resp(conn, opts)
 
+      {:ok, {content_type, body, tracking_fun_data}} ->
+        conn = opts.tracking_fun.(conn, tracking_fun_data)
+
+        send_cached(conn, {content_type, body})
+
       {:ok, record} ->
         send_cached(conn, record)
 
@@ -88,9 +94,17 @@ defmodule Pleroma.Plugs.Cache do
         ttl = Map.get(conn.assigns, :cache_ttl, opts.ttl)
         key = cache_key(conn, opts)
         content_type = content_type(conn)
-        record = {content_type, body}
 
-        Cachex.put(:web_resp_cache, key, record, ttl: ttl)
+        conn =
+          unless opts[:tracking_fun] do
+            Cachex.put(:web_resp_cache, key, {content_type, body}, ttl: ttl)
+            conn
+          else
+            tracking_fun_data = Map.get(conn.assigns, :tracking_fun_data, nil)
+            Cachex.put(:web_resp_cache, key, {content_type, body, tracking_fun_data}, ttl: ttl)
+
+            opts.tracking_fun.(conn, tracking_fun_data)
+          end
 
         put_resp_header(conn, "x-cache", "MISS from Pleroma")
 
diff --git a/lib/pleroma/plugs/http_signature.ex b/lib/pleroma/plugs/http_signature.ex
index d87fa52fa..23d22a712 100644
--- a/lib/pleroma/plugs/http_signature.ex
+++ b/lib/pleroma/plugs/http_signature.ex
@@ -15,7 +15,8 @@ defmodule Pleroma.Web.Plugs.HTTPSignaturePlug do
   end
 
   def call(conn, _opts) do
-    [signature | _] = get_req_header(conn, "signature")
+    headers = get_req_header(conn, "signature")
+    signature = Enum.at(headers, 0)
 
     if signature do
       # set (request-target) header to the appropriate value
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index f0306652c..fb1f24254 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -11,6 +11,7 @@ defmodule Pleroma.User do
   alias Comeonin.Pbkdf2
   alias Ecto.Multi
   alias Pleroma.Activity
+  alias Pleroma.Delivery
   alias Pleroma.Keys
   alias Pleroma.Notification
   alias Pleroma.Object
@@ -62,6 +63,7 @@ defmodule Pleroma.User do
     field(:last_digest_emailed_at, :naive_datetime)
     has_many(:notifications, Notification)
     has_many(:registrations, Registration)
+    has_many(:deliveries, Delivery)
     embeds_one(:info, User.Info)
 
     timestamps()
@@ -148,6 +150,7 @@ defmodule Pleroma.User do
     Cachex.fetch!(:user_cache, key, fn _ -> {:commit, follow_state(user, target)} end)
   end
 
+  @spec set_follow_state_cache(String.t(), String.t(), String.t()) :: {:ok | :error, boolean()}
   def set_follow_state_cache(user_ap_id, target_ap_id, state) do
     Cachex.put(
       :user_cache,
@@ -1640,6 +1643,18 @@ defmodule Pleroma.User do
   def is_internal_user?(%User{local: true, nickname: "internal." <> _}), do: true
   def is_internal_user?(_), do: false
 
+  # A hack because user delete activities have a fake id for whatever reason
+  # TODO: Get rid of this
+  def get_delivered_users_by_object_id("pleroma:fake_object_id"), do: []
+
+  def get_delivered_users_by_object_id(object_id) do
+    from(u in User,
+      inner_join: delivery in assoc(u, :deliveries),
+      where: delivery.object_id == ^object_id
+    )
+    |> Repo.all()
+  end
+
   def change_email(user, email) do
     user
     |> cast(%{email: email}, [:email])
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 41f6a0f1f..e1e90d667 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Activity
+  alias Pleroma.Activity.Ir.Topics
   alias Pleroma.Config
   alias Pleroma.Conversation
   alias Pleroma.Notification
@@ -16,6 +17,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
+  alias Pleroma.Web.Streamer
   alias Pleroma.Web.WebFinger
   alias Pleroma.Workers.BackgroundWorker
 
@@ -187,9 +189,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       participations
       |> Repo.preload(:user)
 
-    Enum.each(participations, fn participation ->
-      Pleroma.Web.Streamer.stream("participation", participation)
-    end)
+    Streamer.stream("participation", participations)
   end
 
   def stream_out_participations(%Object{data: %{"context" => context}}, user) do
@@ -208,41 +208,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   def stream_out_participations(_, _), do: :noop
 
-  def stream_out(activity) do
-    if activity.data["type"] in ["Create", "Announce", "Delete"] do
-      object = Object.normalize(activity)
-      # Do not stream out poll replies
-      unless object.data["type"] == "Answer" do
-        Pleroma.Web.Streamer.stream("user", activity)
-        Pleroma.Web.Streamer.stream("list", activity)
-
-        if get_visibility(activity) == "public" do
-          Pleroma.Web.Streamer.stream("public", activity)
-
-          if activity.local do
-            Pleroma.Web.Streamer.stream("public:local", activity)
-          end
-
-          if activity.data["type"] in ["Create"] do
-            object.data
-            |> Map.get("tag", [])
-            |> Enum.filter(fn tag -> is_bitstring(tag) end)
-            |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
-
-            if object.data["attachment"] != [] do
-              Pleroma.Web.Streamer.stream("public:media", activity)
-
-              if activity.local do
-                Pleroma.Web.Streamer.stream("public:local:media", activity)
-              end
-            end
-          end
-        else
-          if get_visibility(activity) == "direct",
-            do: Pleroma.Web.Streamer.stream("direct", activity)
-        end
-      end
-    end
+  def stream_out(%Activity{data: %{"type" => data_type}} = activity)
+      when data_type in ["Create", "Announce", "Delete"] do
+    activity
+    |> Topics.get_activity_topics()
+    |> Streamer.stream(activity)
+  end
+
+  def stream_out(_activity) do
+    :noop
   end
 
   def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
@@ -436,6 +410,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
+  @spec block(User.t(), User.t(), String.t() | nil, boolean) :: {:ok, Activity.t() | nil}
   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
@@ -464,10 +439,11 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
+  @spec flag(map()) :: {:ok, Activity.t()} | any
   def flag(
         %{
           actor: actor,
-          context: context,
+          context: _context,
           account: account,
           statuses: statuses,
           content: content
@@ -479,14 +455,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
     additional = params[:additional] || %{}
 
-    params = %{
-      actor: actor,
-      context: context,
-      account: account,
-      statuses: statuses,
-      content: content
-    }
-
     additional =
       if forward do
         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
index 705dbc1c2..01b34fb1d 100644
--- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
@@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
   use Pleroma.Web, :controller
 
   alias Pleroma.Activity
+  alias Pleroma.Delivery
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
   alias Pleroma.User
@@ -23,7 +24,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
 
   action_fallback(:errors)
 
-  plug(Pleroma.Plugs.Cache, [query_params: false] when action in [:activity, :object])
+  plug(
+    Pleroma.Plugs.Cache,
+    [query_params: false, tracking_fun: &__MODULE__.track_object_fetch/2]
+    when action in [:activity, :object]
+  )
+
   plug(Pleroma.Web.FederatingPlug when action in [:inbox, :relay])
   plug(:set_requester_reachable when action in [:inbox])
   plug(:relay_active? when action in [:relay])
@@ -54,6 +60,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
          %Object{} = object <- Object.get_cached_by_ap_id(ap_id),
          {_, true} <- {:public?, Visibility.is_public?(object)} do
       conn
+      |> assign(:tracking_fun_data, object.id)
       |> set_cache_ttl_for(object)
       |> put_resp_content_type("application/activity+json")
       |> put_view(ObjectView)
@@ -64,6 +71,16 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     end
   end
 
+  def track_object_fetch(conn, nil), do: conn
+
+  def track_object_fetch(conn, object_id) do
+    with %{assigns: %{user: %User{id: user_id}}} <- conn do
+      Delivery.create(object_id, user_id)
+    end
+
+    conn
+  end
+
   def object_likes(conn, %{"uuid" => uuid, "page" => page}) do
     with ap_id <- o_status_url(conn, :object, uuid),
          %Object{} = object <- Object.get_cached_by_ap_id(ap_id),
@@ -99,6 +116,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
          %Activity{} = activity <- Activity.normalize(ap_id),
          {_, true} <- {:public?, Visibility.is_public?(activity)} do
       conn
+      |> maybe_set_tracking_data(activity)
       |> set_cache_ttl_for(activity)
       |> put_resp_content_type("application/activity+json")
       |> put_view(ObjectView)
@@ -109,6 +127,13 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     end
   end
 
+  defp maybe_set_tracking_data(conn, %Activity{data: %{"type" => "Create"}} = activity) do
+    object_id = Object.normalize(activity).id
+    assign(conn, :tracking_fun_data, object_id)
+  end
+
+  defp maybe_set_tracking_data(conn, _activity), do: conn
+
   defp set_cache_ttl_for(conn, %Activity{object: object}) do
     set_cache_ttl_for(conn, object)
   end
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index a6322e25a..114251b24 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -5,8 +5,10 @@
 defmodule Pleroma.Web.ActivityPub.Publisher do
   alias Pleroma.Activity
   alias Pleroma.Config
+  alias Pleroma.Delivery
   alias Pleroma.HTTP
   alias Pleroma.Instances
+  alias Pleroma.Object
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.Transmogrifier
@@ -116,7 +118,18 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
         {:ok, []}
       end
 
-    Pleroma.Web.Salmon.remote_users(actor, activity) ++ followers
+    fetchers =
+      with %Activity{data: %{"type" => "Delete"}} <- activity,
+           %Object{id: object_id} <- Object.normalize(activity),
+           fetchers <- User.get_delivered_users_by_object_id(object_id),
+           _ <- Delivery.delete_all_by_object_id(object_id) do
+        fetchers
+      else
+        _ ->
+          []
+      end
+
+    Pleroma.Web.Salmon.remote_users(actor, activity) ++ followers ++ fetchers
   end
 
   defp get_cc_ap_ids(ap_id, recipients) do
diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index 2fbb16623..9f6d8b658 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
@@ -1064,7 +1064,7 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
 
   defp upgrade_user(user, data) do
     user
-    |> User.upgrade_changeset(data)
+    |> User.upgrade_changeset(data, true)
     |> User.update_and_set_cache()
   end
 
diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index 258e56066..30628a793 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -33,50 +33,40 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     Map.put(params, "actor", get_ap_id(params["actor"]))
   end
 
-  def determine_explicit_mentions(%{"tag" => tag} = _object) when is_list(tag) do
-    tag
-    |> Enum.filter(fn x -> is_map(x) end)
-    |> Enum.filter(fn x -> x["type"] == "Mention" end)
-    |> Enum.map(fn x -> x["href"] end)
+  @spec determine_explicit_mentions(map()) :: map()
+  def determine_explicit_mentions(%{"tag" => tag} = _) when is_list(tag) do
+    Enum.flat_map(tag, fn
+      %{"type" => "Mention", "href" => href} -> [href]
+      _ -> []
+    end)
   end
 
   def determine_explicit_mentions(%{"tag" => tag} = object) when is_map(tag) do
-    Map.put(object, "tag", [tag])
+    object
+    |> Map.put("tag", [tag])
     |> determine_explicit_mentions()
   end
 
   def determine_explicit_mentions(_), do: []
 
+  @spec recipient_in_collection(any(), any()) :: boolean()
   defp recipient_in_collection(ap_id, coll) when is_binary(coll), do: ap_id == coll
   defp recipient_in_collection(ap_id, coll) when is_list(coll), do: ap_id in coll
   defp recipient_in_collection(_, _), do: false
 
+  @spec recipient_in_message(User.t(), User.t(), map()) :: boolean()
   def recipient_in_message(%User{ap_id: ap_id} = recipient, %User{} = actor, params) do
-    cond do
-      recipient_in_collection(ap_id, params["to"]) ->
-        true
-
-      recipient_in_collection(ap_id, params["cc"]) ->
-        true
-
-      recipient_in_collection(ap_id, params["bto"]) ->
-        true
-
-      recipient_in_collection(ap_id, params["bcc"]) ->
-        true
+    addresses = [params["to"], params["cc"], params["bto"], params["bcc"]]
 
+    cond do
+      Enum.any?(addresses, &recipient_in_collection(ap_id, &1)) -> true
       # if the message is unaddressed at all, then assume it is directly addressed
       # to the recipient
-      !params["to"] && !params["cc"] && !params["bto"] && !params["bcc"] ->
-        true
-
+      Enum.all?(addresses, &is_nil(&1)) -> true
       # if the message is sent from somebody the user is following, then assume it
       # is addressed to the recipient
-      User.following?(recipient, actor) ->
-        true
-
-      true ->
-        false
+      User.following?(recipient, actor) -> true
+      true -> false
     end
   end
 
@@ -179,54 +169,59 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   Adds an id and a published data if they aren't there,
   also adds it to an included object
   """
-  def lazy_put_activity_defaults(map, fake? \\ false) do
-    map =
-      if not fake? do
-        %{data: %{"id" => context}, id: context_id} = create_context(map["context"])
-
-        map
-        |> Map.put_new_lazy("id", &generate_activity_id/0)
-        |> Map.put_new_lazy("published", &make_date/0)
-        |> Map.put_new("context", context)
-        |> Map.put_new("context_id", context_id)
-      else
-        map
-        |> Map.put_new("id", "pleroma:fakeid")
-        |> Map.put_new_lazy("published", &make_date/0)
-        |> Map.put_new("context", "pleroma:fakecontext")
-        |> Map.put_new("context_id", -1)
-      end
+  @spec lazy_put_activity_defaults(map(), boolean) :: map()
+  def lazy_put_activity_defaults(map, fake? \\ false)
 
-    if is_map(map["object"]) do
-      object = lazy_put_object_defaults(map["object"], map, fake?)
-      %{map | "object" => object}
-    else
-      map
-    end
+  def lazy_put_activity_defaults(map, true) do
+    map
+    |> Map.put_new("id", "pleroma:fakeid")
+    |> Map.put_new_lazy("published", &make_date/0)
+    |> Map.put_new("context", "pleroma:fakecontext")
+    |> Map.put_new("context_id", -1)
+    |> lazy_put_object_defaults(true)
   end
 
-  @doc """
-  Adds an id and published date if they aren't there.
-  """
-  def lazy_put_object_defaults(map, activity \\ %{}, fake?)
+  def lazy_put_activity_defaults(map, _fake?) do
+    %{data: %{"id" => context}, id: context_id} = create_context(map["context"])
 
-  def lazy_put_object_defaults(map, activity, true = _fake?) do
     map
+    |> Map.put_new_lazy("id", &generate_activity_id/0)
     |> Map.put_new_lazy("published", &make_date/0)
-    |> Map.put_new("id", "pleroma:fake_object_id")
-    |> Map.put_new("context", activity["context"])
-    |> Map.put_new("fake", true)
-    |> Map.put_new("context_id", activity["context_id"])
+    |> Map.put_new("context", context)
+    |> Map.put_new("context_id", context_id)
+    |> lazy_put_object_defaults(false)
   end
 
-  def lazy_put_object_defaults(map, activity, _fake?) do
-    map
-    |> Map.put_new_lazy("id", &generate_object_id/0)
-    |> Map.put_new_lazy("published", &make_date/0)
-    |> Map.put_new("context", activity["context"])
-    |> Map.put_new("context_id", activity["context_id"])
+  # Adds an id and published date if they aren't there.
+  #
+  @spec lazy_put_object_defaults(map(), boolean()) :: map()
+  defp lazy_put_object_defaults(%{"object" => map} = activity, true)
+       when is_map(map) do
+    object =
+      map
+      |> Map.put_new("id", "pleroma:fake_object_id")
+      |> Map.put_new_lazy("published", &make_date/0)
+      |> Map.put_new("context", activity["context"])
+      |> Map.put_new("context_id", activity["context_id"])
+      |> Map.put_new("fake", true)
+
+    %{activity | "object" => object}
+  end
+
+  defp lazy_put_object_defaults(%{"object" => map} = activity, _)
+       when is_map(map) do
+    object =
+      map
+      |> Map.put_new_lazy("id", &generate_object_id/0)
+      |> Map.put_new_lazy("published", &make_date/0)
+      |> Map.put_new("context", activity["context"])
+      |> Map.put_new("context_id", activity["context_id"])
+
+    %{activity | "object" => object}
   end
 
+  defp lazy_put_object_defaults(activity, _), do: activity
+
   @doc """
   Inserts a full object if it is contained in an activity.
   """
@@ -345,24 +340,24 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   @doc """
   Updates a follow activity's state (for locked accounts).
   """
+  @spec update_follow_state_for_all(Activity.t(), String.t()) :: {:ok, Activity} | {:error, any()}
   def update_follow_state_for_all(
         %Activity{data: %{"actor" => actor, "object" => object}} = activity,
         state
       ) do
-    try do
-      Ecto.Adapters.SQL.query!(
-        Repo,
-        "UPDATE activities SET data = jsonb_set(data, '{state}', $1) WHERE data->>'type' = 'Follow' AND data->>'actor' = $2 AND data->>'object' = $3 AND data->>'state' = 'pending'",
-        [state, actor, object]
-      )
+    "Follow"
+    |> Activity.Queries.by_type()
+    |> Activity.Queries.by_actor(actor)
+    |> Activity.Queries.by_object_id(object)
+    |> where(fragment("data->>'state' = 'pending'"))
+    |> update(set: [data: fragment("jsonb_set(data, '{state}', ?)", ^state)])
+    |> Repo.update_all([])
 
-      User.set_follow_state_cache(actor, object, state)
-      activity = Activity.get_by_id(activity.id)
-      {:ok, activity}
-    rescue
-      e ->
-        {:error, e}
-    end
+    User.set_follow_state_cache(actor, object, state)
+
+    activity = Activity.get_by_id(activity.id)
+
+    {:ok, activity}
   end
 
   def update_follow_state(
@@ -413,6 +408,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   @doc """
   Retruns an existing announce activity if the notice has already been announced
   """
+  @spec get_existing_announce(String.t(), map()) :: Activity.t() | nil
   def get_existing_announce(actor, %{data: %{"id" => ap_id}}) do
     "Announce"
     |> Activity.Queries.by_type()
@@ -495,33 +491,35 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     |> maybe_put("id", activity_id)
   end
 
+  @spec add_announce_to_object(Activity.t(), Object.t()) ::
+          {:ok, Object.t()} | {:error, Ecto.Changeset.t()}
   def add_announce_to_object(
-        %Activity{
-          data: %{"actor" => actor, "cc" => [Pleroma.Constants.as_public()]}
-        },
+        %Activity{data: %{"actor" => actor, "cc" => [Pleroma.Constants.as_public()]}},
         object
       ) do
-    announcements =
-      if is_list(object.data["announcements"]) do
-        Enum.uniq([actor | object.data["announcements"]])
-      else
-        [actor]
-      end
+    announcements = take_announcements(object)
 
-    update_element_in_object("announcement", announcements, object)
+    with announcements <- Enum.uniq([actor | announcements]) do
+      update_element_in_object("announcement", announcements, object)
+    end
   end
 
   def add_announce_to_object(_, object), do: {:ok, object}
 
+  @spec remove_announce_from_object(Activity.t(), Object.t()) ::
+          {:ok, Object.t()} | {:error, Ecto.Changeset.t()}
   def remove_announce_from_object(%Activity{data: %{"actor" => actor}}, object) do
-    announcements =
-      if is_list(object.data["announcements"]), do: object.data["announcements"], else: []
-
-    with announcements <- announcements |> List.delete(actor) do
+    with announcements <- List.delete(take_announcements(object), actor) do
       update_element_in_object("announcement", announcements, object)
     end
   end
 
+  defp take_announcements(%{data: %{"announcements" => announcements}} = _)
+       when is_list(announcements),
+       do: announcements
+
+  defp take_announcements(_), do: []
+
   #### Unfollow-related helpers
 
   def make_unfollow_data(follower, followed, follow_activity, activity_id) do
@@ -535,6 +533,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   end
 
   #### Block-related helpers
+  @spec fetch_latest_block(User.t(), User.t()) :: Activity.t() | nil
   def fetch_latest_block(%User{ap_id: blocker_id}, %User{ap_id: blocked_id}) do
     "Block"
     |> Activity.Queries.by_type()
@@ -583,28 +582,32 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   end
 
   #### Flag-related helpers
-
-  def make_flag_data(params, additional) do
-    status_ap_ids =
-      Enum.map(params.statuses || [], fn
-        %Activity{} = act -> act.data["id"]
-        act when is_map(act) -> act["id"]
-        act when is_binary(act) -> act
-      end)
-
-    object = [params.account.ap_id] ++ status_ap_ids
-
+  @spec make_flag_data(map(), map()) :: map()
+  def make_flag_data(%{actor: actor, context: context, content: content} = params, additional) do
     %{
       "type" => "Flag",
-      "actor" => params.actor.ap_id,
-      "content" => params.content,
-      "object" => object,
-      "context" => params.context,
+      "actor" => actor.ap_id,
+      "content" => content,
+      "object" => build_flag_object(params),
+      "context" => context,
       "state" => "open"
     }
     |> Map.merge(additional)
   end
 
+  def make_flag_data(_, _), do: %{}
+
+  defp build_flag_object(%{account: account, statuses: statuses} = _) do
+    [account.ap_id] ++
+      Enum.map(statuses || [], fn
+        %Activity{} = act -> act.data["id"]
+        act when is_map(act) -> act["id"]
+        act when is_binary(act) -> act
+      end)
+  end
+
+  defp build_flag_object(_), do: []
+
   @doc """
   Fetches the OrderedCollection/OrderedCollectionPage from `from`, limiting the amount of pages fetched after
   the first one to `pages_left` pages.
diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
index dbd3542ea..3c26eb406 100644
--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
@@ -8,6 +8,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   alias Pleroma.Repo
   alias Pleroma.User
   alias Pleroma.Web.OAuth.Token
+  alias Pleroma.Web.Streamer
 
   @behaviour :cowboy_websocket
 
@@ -24,7 +25,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   ]
   @anonymous_streams ["public", "public:local", "hashtag"]
 
-  # Handled by periodic keepalive in Pleroma.Web.Streamer.
+  # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
   @timeout :infinity
 
   def init(%{qs: qs} = req, state) do
@@ -65,7 +66,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
       }, topic #{state.topic}"
     )
 
-    Pleroma.Web.Streamer.add_socket(state.topic, streamer_socket(state))
+    Streamer.add_socket(state.topic, streamer_socket(state))
     {:ok, state}
   end
 
@@ -80,7 +81,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
       }, topic #{state.topic || "?"}: #{inspect(reason)}"
     )
 
-    Pleroma.Web.Streamer.remove_socket(state.topic, streamer_socket(state))
+    Streamer.remove_socket(state.topic, streamer_socket(state))
     :ok
   end
 
diff --git a/lib/pleroma/web/rich_media/parser.ex b/lib/pleroma/web/rich_media/parser.ex
index f5f9e358c..c06b0a0f2 100644
--- a/lib/pleroma/web/rich_media/parser.ex
+++ b/lib/pleroma/web/rich_media/parser.ex
@@ -81,6 +81,7 @@ defmodule Pleroma.Web.RichMedia.Parser do
       {:ok, %Tesla.Env{body: html}} = Pleroma.HTTP.get(url, [], adapter: @hackney_options)
 
       html
+      |> parse_html
       |> maybe_parse()
       |> Map.put(:url, url)
       |> clean_parsed_data()
@@ -91,6 +92,8 @@ defmodule Pleroma.Web.RichMedia.Parser do
     end
   end
 
+  defp parse_html(html), do: Floki.parse(html)
+
   defp maybe_parse(html) do
     Enum.reduce_while(parsers(), %{}, fn parser, acc ->
       case parser.parse(html, acc) do
@@ -100,7 +103,8 @@ defmodule Pleroma.Web.RichMedia.Parser do
     end)
   end
 
-  defp check_parsed_data(%{title: title} = data) when is_binary(title) and byte_size(title) > 0 do
+  defp check_parsed_data(%{title: title} = data)
+       when is_binary(title) and byte_size(title) > 0 do
     {:ok, data}
   end
 
diff --git a/lib/pleroma/web/router.ex b/lib/pleroma/web/router.ex
index b0464037e..401133bf3 100644
--- a/lib/pleroma/web/router.ex
+++ b/lib/pleroma/web/router.ex
@@ -135,6 +135,7 @@ defmodule Pleroma.Web.Router do
 
   pipeline :http_signature do
     plug(Pleroma.Web.Plugs.HTTPSignaturePlug)
+    plug(Pleroma.Web.Plugs.MappedSignatureToIdentityPlug)
   end
 
   scope "/api/pleroma", Pleroma.Web.TwitterAPI do
@@ -514,6 +515,7 @@ defmodule Pleroma.Web.Router do
 
   scope "/", Pleroma.Web do
     pipe_through(:ostatus)
+    pipe_through(:http_signature)
 
     get("/objects/:uuid", OStatus.OStatusController, :object)
     get("/activities/:uuid", OStatus.OStatusController, :activity)
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
deleted file mode 100644
index 587c43f40..000000000
--- a/lib/pleroma/web/streamer.ex
+++ /dev/null
@@ -1,318 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Streamer do
-  use GenServer
-  require Logger
-  alias Pleroma.Activity
-  alias Pleroma.Config
-  alias Pleroma.Conversation.Participation
-  alias Pleroma.Notification
-  alias Pleroma.Object
-  alias Pleroma.User
-  alias Pleroma.Web.ActivityPub.ActivityPub
-  alias Pleroma.Web.ActivityPub.Visibility
-  alias Pleroma.Web.CommonAPI
-  alias Pleroma.Web.MastodonAPI.NotificationView
-
-  @keepalive_interval :timer.seconds(30)
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
-  end
-
-  def add_socket(topic, socket) do
-    GenServer.cast(__MODULE__, %{action: :add, socket: socket, topic: topic})
-  end
-
-  def remove_socket(topic, socket) do
-    GenServer.cast(__MODULE__, %{action: :remove, socket: socket, topic: topic})
-  end
-
-  def stream(topic, item) do
-    GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item})
-  end
-
-  def init(args) do
-    Process.send_after(self(), %{action: :ping}, @keepalive_interval)
-
-    {:ok, args}
-  end
-
-  def handle_info(%{action: :ping}, topics) do
-    topics
-    |> Map.values()
-    |> List.flatten()
-    |> Enum.each(fn socket ->
-      Logger.debug("Sending keepalive ping")
-      send(socket.transport_pid, {:text, ""})
-    end)
-
-    Process.send_after(self(), %{action: :ping}, @keepalive_interval)
-
-    {:noreply, topics}
-  end
-
-  def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
-    recipient_topics =
-      User.get_recipients_from_activity(item)
-      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
-
-    Enum.each(recipient_topics || [], fn user_topic ->
-      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
-      push_to_socket(topics, user_topic, item)
-    end)
-
-    {:noreply, topics}
-  end
-
-  def handle_cast(%{action: :stream, topic: "participation", item: participation}, topics) do
-    user_topic = "direct:#{participation.user_id}"
-    Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
-
-    push_to_socket(topics, user_topic, participation)
-
-    {:noreply, topics}
-  end
-
-  def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
-    # filter the recipient list if the activity is not public, see #270.
-    recipient_lists =
-      case Visibility.is_public?(item) do
-        true ->
-          Pleroma.List.get_lists_from_activity(item)
-
-        _ ->
-          Pleroma.List.get_lists_from_activity(item)
-          |> Enum.filter(fn list ->
-            owner = User.get_cached_by_id(list.user_id)
-
-            Visibility.visible_for_user?(item, owner)
-          end)
-      end
-
-    recipient_topics =
-      recipient_lists
-      |> Enum.map(fn %{id: id} -> "list:#{id}" end)
-
-    Enum.each(recipient_topics || [], fn list_topic ->
-      Logger.debug("Trying to push message to #{list_topic}\n\n")
-      push_to_socket(topics, list_topic, item)
-    end)
-
-    {:noreply, topics}
-  end
-
-  def handle_cast(
-        %{action: :stream, topic: topic, item: %Notification{} = item},
-        topics
-      )
-      when topic in ["user", "user:notification"] do
-    topics
-    |> Map.get("#{topic}:#{item.user_id}", [])
-    |> Enum.each(fn socket ->
-      with %User{} = user <- User.get_cached_by_ap_id(socket.assigns[:user].ap_id),
-           true <- should_send?(user, item) do
-        send(
-          socket.transport_pid,
-          {:text, represent_notification(socket.assigns[:user], item)}
-        )
-      end
-    end)
-
-    {:noreply, topics}
-  end
-
-  def handle_cast(%{action: :stream, topic: "user", item: item}, topics) do
-    Logger.debug("Trying to push to users")
-
-    recipient_topics =
-      User.get_recipients_from_activity(item)
-      |> Enum.map(fn %{id: id} -> "user:#{id}" end)
-
-    Enum.each(recipient_topics, fn topic ->
-      push_to_socket(topics, topic, item)
-    end)
-
-    {:noreply, topics}
-  end
-
-  def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do
-    Logger.debug("Trying to push to #{topic}")
-    Logger.debug("Pushing item to #{topic}")
-    push_to_socket(topics, topic, item)
-    {:noreply, topics}
-  end
-
-  def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do
-    topic = internal_topic(topic, socket)
-    sockets_for_topic = sockets[topic] || []
-    sockets_for_topic = Enum.uniq([socket | sockets_for_topic])
-    sockets = Map.put(sockets, topic, sockets_for_topic)
-    Logger.debug("Got new conn for #{topic}")
-    {:noreply, sockets}
-  end
-
-  def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do
-    topic = internal_topic(topic, socket)
-    sockets_for_topic = sockets[topic] || []
-    sockets_for_topic = List.delete(sockets_for_topic, socket)
-    sockets = Map.put(sockets, topic, sockets_for_topic)
-    Logger.debug("Removed conn for #{topic}")
-    {:noreply, sockets}
-  end
-
-  def handle_cast(m, state) do
-    Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}")
-    {:noreply, state}
-  end
-
-  defp represent_update(%Activity{} = activity, %User{} = user) do
-    %{
-      event: "update",
-      payload:
-        Pleroma.Web.MastodonAPI.StatusView.render(
-          "status.json",
-          activity: activity,
-          for: user
-        )
-        |> Jason.encode!()
-    }
-    |> Jason.encode!()
-  end
-
-  defp represent_update(%Activity{} = activity) do
-    %{
-      event: "update",
-      payload:
-        Pleroma.Web.MastodonAPI.StatusView.render(
-          "status.json",
-          activity: activity
-        )
-        |> Jason.encode!()
-    }
-    |> Jason.encode!()
-  end
-
-  def represent_conversation(%Participation{} = participation) do
-    %{
-      event: "conversation",
-      payload:
-        Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
-          participation: participation,
-          for: participation.user
-        })
-        |> Jason.encode!()
-    }
-    |> Jason.encode!()
-  end
-
-  @spec represent_notification(User.t(), Notification.t()) :: binary()
-  defp represent_notification(%User{} = user, %Notification{} = notify) do
-    %{
-      event: "notification",
-      payload:
-        NotificationView.render(
-          "show.json",
-          %{notification: notify, for: user}
-        )
-        |> Jason.encode!()
-    }
-    |> Jason.encode!()
-  end
-
-  defp should_send?(%User{} = user, %Activity{} = item) do
-    blocks = user.info.blocks || []
-    mutes = user.info.mutes || []
-    reblog_mutes = user.info.muted_reblogs || []
-    domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)
-
-    with parent when not is_nil(parent) <- Object.normalize(item),
-         true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
-         true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
-         %{host: item_host} <- URI.parse(item.actor),
-         %{host: parent_host} <- URI.parse(parent.data["actor"]),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
-         true <- thread_containment(item, user),
-         false <- CommonAPI.thread_muted?(user, item) do
-      true
-    else
-      _ -> false
-    end
-  end
-
-  defp should_send?(%User{} = user, %Notification{activity: activity}) do
-    should_send?(user, activity)
-  end
-
-  def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
-    Enum.each(topics[topic] || [], fn socket ->
-      # Get the current user so we have up-to-date blocks etc.
-      if socket.assigns[:user] do
-        user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
-
-        if should_send?(user, item) do
-          send(socket.transport_pid, {:text, represent_update(item, user)})
-        end
-      else
-        send(socket.transport_pid, {:text, represent_update(item)})
-      end
-    end)
-  end
-
-  def push_to_socket(topics, topic, %Participation{} = participation) do
-    Enum.each(topics[topic] || [], fn socket ->
-      send(socket.transport_pid, {:text, represent_conversation(participation)})
-    end)
-  end
-
-  def push_to_socket(topics, topic, %Activity{
-        data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
-      }) do
-    Enum.each(topics[topic] || [], fn socket ->
-      send(
-        socket.transport_pid,
-        {:text, %{event: "delete", payload: to_string(deleted_activity_id)} |> Jason.encode!()}
-      )
-    end)
-  end
-
-  def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
-
-  def push_to_socket(topics, topic, item) do
-    Enum.each(topics[topic] || [], fn socket ->
-      # Get the current user so we have up-to-date blocks etc.
-      if socket.assigns[:user] do
-        user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
-        blocks = user.info.blocks || []
-        mutes = user.info.mutes || []
-
-        with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
-             true <- thread_containment(item, user) do
-          send(socket.transport_pid, {:text, represent_update(item, user)})
-        end
-      else
-        send(socket.transport_pid, {:text, represent_update(item)})
-      end
-    end)
-  end
-
-  defp internal_topic(topic, socket) when topic in ~w[user user:notification direct] do
-    "#{topic}:#{socket.assigns[:user].id}"
-  end
-
-  defp internal_topic(topic, _), do: topic
-
-  @spec thread_containment(Activity.t(), User.t()) :: boolean()
-  defp thread_containment(_activity, %User{info: %{skip_thread_containment: true}}), do: true
-
-  defp thread_containment(activity, user) do
-    if Config.get([:instance, :skip_thread_containment]) do
-      true
-    else
-      ActivityPub.contain_activity(activity, user)
-    end
-  end
-end
diff --git a/lib/pleroma/web/streamer/ping.ex b/lib/pleroma/web/streamer/ping.ex
new file mode 100644
index 000000000..f77cbb95c
--- /dev/null
+++ b/lib/pleroma/web/streamer/ping.ex
@@ -0,0 +1,33 @@
+defmodule Pleroma.Web.Streamer.Ping do
+  use GenServer
+  require Logger
+
+  alias Pleroma.Web.Streamer.State
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  @keepalive_interval :timer.seconds(30)
+
+  def start_link(opts) do
+    ping_interval = Keyword.get(opts, :ping_interval, @keepalive_interval)
+    GenServer.start_link(__MODULE__, %{ping_interval: ping_interval}, name: __MODULE__)
+  end
+
+  def init(%{ping_interval: ping_interval} = args) do
+    Process.send_after(self(), :ping, ping_interval)
+    {:ok, args}
+  end
+
+  def handle_info(:ping, %{ping_interval: ping_interval} = state) do
+    State.get_sockets()
+    |> Map.values()
+    |> List.flatten()
+    |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid} ->
+      Logger.debug("Sending keepalive ping")
+      send(transport_pid, {:text, ""})
+    end)
+
+    Process.send_after(self(), :ping, ping_interval)
+
+    {:noreply, state}
+  end
+end
diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex
new file mode 100644
index 000000000..c48752d95
--- /dev/null
+++ b/lib/pleroma/web/streamer/state.ex
@@ -0,0 +1,78 @@
+defmodule Pleroma.Web.Streamer.State do
+  use GenServer
+  require Logger
+
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  @env Mix.env()
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
+  end
+
+  def add_socket(topic, socket) do
+    GenServer.call(__MODULE__, {:add, topic, socket})
+  end
+
+  def remove_socket(topic, socket) do
+    do_remove_socket(@env, topic, socket)
+  end
+
+  def get_sockets do
+    %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
+    stream_sockets
+  end
+
+  def init(init_arg) do
+    {:ok, init_arg}
+  end
+
+  def handle_call(:get_state, _from, state) do
+    {:reply, state, state}
+  end
+
+  def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
+    internal_topic = internal_topic(topic, socket)
+    stream_socket = StreamerSocket.from_socket(socket)
+
+    sockets_for_topic =
+      sockets
+      |> Map.get(internal_topic, [])
+      |> List.insert_at(0, stream_socket)
+      |> Enum.uniq()
+
+    state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
+    Logger.debug("Got new conn for #{topic}")
+    {:reply, state, state}
+  end
+
+  def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
+    internal_topic = internal_topic(topic, socket)
+    stream_socket = StreamerSocket.from_socket(socket)
+
+    sockets_for_topic =
+      sockets
+      |> Map.get(internal_topic, [])
+      |> List.delete(stream_socket)
+
+    state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
+    {:reply, state, state}
+  end
+
+  defp do_remove_socket(:test, _, _) do
+    :ok
+  end
+
+  defp do_remove_socket(_env, topic, socket) do
+    GenServer.call(__MODULE__, {:remove, topic, socket})
+  end
+
+  defp internal_topic(topic, socket)
+       when topic in ~w[user user:notification direct] do
+    "#{topic}:#{socket.assigns[:user].id}"
+  end
+
+  defp internal_topic(topic, _) do
+    topic
+  end
+end
diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex
new file mode 100644
index 000000000..8cf719277
--- /dev/null
+++ b/lib/pleroma/web/streamer/streamer.ex
@@ -0,0 +1,55 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.Streamer do
+  alias Pleroma.Web.Streamer.State
+  alias Pleroma.Web.Streamer.Worker
+
+  @timeout 60_000
+  @mix_env Mix.env()
+
+  def add_socket(topic, socket) do
+    State.add_socket(topic, socket)
+  end
+
+  def remove_socket(topic, socket) do
+    State.remove_socket(topic, socket)
+  end
+
+  def get_sockets do
+    State.get_sockets()
+  end
+
+  def stream(topics, items) do
+    if should_send?() do
+      Task.async(fn ->
+        :poolboy.transaction(
+          :streamer_worker,
+          &Worker.stream(&1, topics, items),
+          @timeout
+        )
+      end)
+    end
+  end
+
+  def supervisor, do: Pleroma.Web.Streamer.Supervisor
+
+  defp should_send? do
+    handle_should_send(@mix_env)
+  end
+
+  defp handle_should_send(:test) do
+    case Process.whereis(:streamer_worker) do
+      nil ->
+        false
+
+      pid ->
+        Process.alive?(pid)
+    end
+  end
+
+  defp handle_should_send(_) do
+    true
+  end
+end
diff --git a/lib/pleroma/web/streamer/streamer_socket.ex b/lib/pleroma/web/streamer/streamer_socket.ex
new file mode 100644
index 000000000..f006c0306
--- /dev/null
+++ b/lib/pleroma/web/streamer/streamer_socket.ex
@@ -0,0 +1,31 @@
+defmodule Pleroma.Web.Streamer.StreamerSocket do
+  defstruct transport_pid: nil, user: nil
+
+  alias Pleroma.User
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  def from_socket(%{
+        transport_pid: transport_pid,
+        assigns: %{user: nil}
+      }) do
+    %StreamerSocket{
+      transport_pid: transport_pid
+    }
+  end
+
+  def from_socket(%{
+        transport_pid: transport_pid,
+        assigns: %{user: %User{} = user}
+      }) do
+    %StreamerSocket{
+      transport_pid: transport_pid,
+      user: user
+    }
+  end
+
+  def from_socket(%{transport_pid: transport_pid}) do
+    %StreamerSocket{
+      transport_pid: transport_pid
+    }
+  end
+end
diff --git a/lib/pleroma/web/streamer/supervisor.ex b/lib/pleroma/web/streamer/supervisor.ex
new file mode 100644
index 000000000..6afe19323
--- /dev/null
+++ b/lib/pleroma/web/streamer/supervisor.ex
@@ -0,0 +1,33 @@
+defmodule Pleroma.Web.Streamer.Supervisor do
+  use Supervisor
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def init(args) do
+    children = [
+      {Pleroma.Web.Streamer.State, args},
+      {Pleroma.Web.Streamer.Ping, args},
+      :poolboy.child_spec(:streamer_worker, poolboy_config())
+    ]
+
+    opts = [strategy: :one_for_one, name: Pleroma.Web.Streamer.Supervisor]
+    Supervisor.init(children, opts)
+  end
+
+  defp poolboy_config do
+    opts =
+      Pleroma.Config.get(:streamer,
+        workers: 3,
+        overflow_workers: 2
+      )
+
+    [
+      {:name, {:local, :streamer_worker}},
+      {:worker_module, Pleroma.Web.Streamer.Worker},
+      {:size, opts[:workers]},
+      {:max_overflow, opts[:overflow_workers]}
+    ]
+  end
+end
diff --git a/lib/pleroma/web/streamer/worker.ex b/lib/pleroma/web/streamer/worker.ex
new file mode 100644
index 000000000..5804508eb
--- /dev/null
+++ b/lib/pleroma/web/streamer/worker.ex
@@ -0,0 +1,220 @@
+defmodule Pleroma.Web.Streamer.Worker do
+  use GenServer
+
+  require Logger
+
+  alias Pleroma.Activity
+  alias Pleroma.Config
+  alias Pleroma.Conversation.Participation
+  alias Pleroma.Notification
+  alias Pleroma.Object
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.Visibility
+  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Web.Streamer.State
+  alias Pleroma.Web.Streamer.StreamerSocket
+  alias Pleroma.Web.StreamerView
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, %{}, [])
+  end
+
+  def init(init_arg) do
+    {:ok, init_arg}
+  end
+
+  def stream(pid, topics, items) do
+    GenServer.call(pid, {:stream, topics, items})
+  end
+
+  def handle_call({:stream, topics, item}, _from, state) when is_list(topics) do
+    Enum.each(topics, fn t ->
+      do_stream(%{topic: t, item: item})
+    end)
+
+    {:reply, state, state}
+  end
+
+  def handle_call({:stream, topic, items}, _from, state) when is_list(items) do
+    Enum.each(items, fn i ->
+      do_stream(%{topic: topic, item: i})
+    end)
+
+    {:reply, state, state}
+  end
+
+  def handle_call({:stream, topic, item}, _from, state) do
+    do_stream(%{topic: topic, item: item})
+
+    {:reply, state, state}
+  end
+
+  defp do_stream(%{topic: "direct", item: item}) do
+    recipient_topics =
+      User.get_recipients_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
+
+    Enum.each(recipient_topics, fn user_topic ->
+      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
+      push_to_socket(State.get_sockets(), user_topic, item)
+    end)
+  end
+
+  defp do_stream(%{topic: "participation", item: participation}) do
+    user_topic = "direct:#{participation.user_id}"
+    Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
+
+    push_to_socket(State.get_sockets(), user_topic, participation)
+  end
+
+  defp do_stream(%{topic: "list", item: item}) do
+    # filter the recipient list if the activity is not public, see #270.
+    recipient_lists =
+      case Visibility.is_public?(item) do
+        true ->
+          Pleroma.List.get_lists_from_activity(item)
+
+        _ ->
+          Pleroma.List.get_lists_from_activity(item)
+          |> Enum.filter(fn list ->
+            owner = User.get_cached_by_id(list.user_id)
+
+            Visibility.visible_for_user?(item, owner)
+          end)
+      end
+
+    recipient_topics =
+      recipient_lists
+      |> Enum.map(fn %{id: id} -> "list:#{id}" end)
+
+    Enum.each(recipient_topics, fn list_topic ->
+      Logger.debug("Trying to push message to #{list_topic}\n\n")
+      push_to_socket(State.get_sockets(), list_topic, item)
+    end)
+  end
+
+  defp do_stream(%{topic: topic, item: %Notification{} = item})
+       when topic in ["user", "user:notification"] do
+    State.get_sockets()
+    |> Map.get("#{topic}:#{item.user_id}", [])
+    |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} ->
+      with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id),
+           true <- should_send?(user, item) do
+        send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)})
+      end
+    end)
+  end
+
+  defp do_stream(%{topic: "user", item: item}) do
+    Logger.debug("Trying to push to users")
+
+    recipient_topics =
+      User.get_recipients_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "user:#{id}" end)
+
+    Enum.each(recipient_topics, fn topic ->
+      push_to_socket(State.get_sockets(), topic, item)
+    end)
+  end
+
+  defp do_stream(%{topic: topic, item: item}) do
+    Logger.debug("Trying to push to #{topic}")
+    Logger.debug("Pushing item to #{topic}")
+    push_to_socket(State.get_sockets(), topic, item)
+  end
+
+  defp should_send?(%User{} = user, %Activity{} = item) do
+    blocks = user.info.blocks || []
+    mutes = user.info.mutes || []
+    reblog_mutes = user.info.muted_reblogs || []
+    domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)
+
+    with parent when not is_nil(parent) <- Object.normalize(item),
+         true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
+         true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
+         %{host: item_host} <- URI.parse(item.actor),
+         %{host: parent_host} <- URI.parse(parent.data["actor"]),
+         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
+         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
+         true <- thread_containment(item, user),
+         false <- CommonAPI.thread_muted?(user, item) do
+      true
+    else
+      _ -> false
+    end
+  end
+
+  defp should_send?(%User{} = user, %Notification{activity: activity}) do
+    should_send?(user, activity)
+  end
+
+  def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
+    Enum.each(topics[topic] || [], fn %StreamerSocket{
+                                        transport_pid: transport_pid,
+                                        user: socket_user
+                                      } ->
+      # Get the current user so we have up-to-date blocks etc.
+      if socket_user do
+        user = User.get_cached_by_ap_id(socket_user.ap_id)
+
+        if should_send?(user, item) do
+          send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
+        end
+      else
+        send(transport_pid, {:text, StreamerView.render("update.json", item)})
+      end
+    end)
+  end
+
+  def push_to_socket(topics, topic, %Participation{} = participation) do
+    Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} ->
+      send(transport_pid, {:text, StreamerView.render("conversation.json", participation)})
+    end)
+  end
+
+  def push_to_socket(topics, topic, %Activity{
+        data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
+      }) do
+    Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} ->
+      send(
+        transport_pid,
+        {:text, %{event: "delete", payload: to_string(deleted_activity_id)} |> Jason.encode!()}
+      )
+    end)
+  end
+
+  def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
+
+  def push_to_socket(topics, topic, item) do
+    Enum.each(topics[topic] || [], fn %StreamerSocket{
+                                        transport_pid: transport_pid,
+                                        user: socket_user
+                                      } ->
+      # Get the current user so we have up-to-date blocks etc.
+      if socket_user do
+        user = User.get_cached_by_ap_id(socket_user.ap_id)
+        blocks = user.info.blocks || []
+        mutes = user.info.mutes || []
+
+        with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
+             true <- thread_containment(item, user) do
+          send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
+        end
+      else
+        send(transport_pid, {:text, StreamerView.render("update.json", item)})
+      end
+    end)
+  end
+
+  @spec thread_containment(Activity.t(), User.t()) :: boolean()
+  defp thread_containment(_activity, %User{info: %{skip_thread_containment: true}}), do: true
+
+  defp thread_containment(activity, user) do
+    if Config.get([:instance, :skip_thread_containment]) do
+      true
+    else
+      ActivityPub.contain_activity(activity, user)
+    end
+  end
+end
diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex
new file mode 100644
index 000000000..b13030fa0
--- /dev/null
+++ b/lib/pleroma/web/views/streamer_view.ex
@@ -0,0 +1,66 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.StreamerView do
+  use Pleroma.Web, :view
+
+  alias Pleroma.Activity
+  alias Pleroma.Conversation.Participation
+  alias Pleroma.Notification
+  alias Pleroma.User
+  alias Pleroma.Web.MastodonAPI.NotificationView
+
+  def render("update.json", %Activity{} = activity, %User{} = user) do
+    %{
+      event: "update",
+      payload:
+        Pleroma.Web.MastodonAPI.StatusView.render(
+          "status.json",
+          activity: activity,
+          for: user
+        )
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
+  def render("notification.json", %User{} = user, %Notification{} = notify) do
+    %{
+      event: "notification",
+      payload:
+        NotificationView.render(
+          "show.json",
+          %{notification: notify, for: user}
+        )
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
+  def render("update.json", %Activity{} = activity) do
+    %{
+      event: "update",
+      payload:
+        Pleroma.Web.MastodonAPI.StatusView.render(
+          "status.json",
+          activity: activity
+        )
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
+  def render("conversation.json", %Participation{} = participation) do
+    %{
+      event: "conversation",
+      payload:
+        Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
+          participation: participation,
+          for: participation.user
+        })
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+end
diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex
index bea2baffb..61b451e3e 100644
--- a/lib/pleroma/workers/web_pusher_worker.ex
+++ b/lib/pleroma/workers/web_pusher_worker.ex
@@ -10,7 +10,11 @@ defmodule Pleroma.Workers.WebPusherWorker do
 
   @impl Oban.Worker
   def perform(%{"op" => "web_push", "notification_id" => notification_id}, _job) do
-    notification = Repo.get(Notification, notification_id)
+    notification =
+      Notification
+      |> Repo.get(notification_id)
+      |> Repo.preload([:activity])
+
     Pleroma.Web.Push.Impl.perform(notification)
   end
 end