From 8cbad5500cefbba1e0bb67604960fc331b75b498 Mon Sep 17 00:00:00 2001
From: Maksim Pechnikov <parallel588@gmail.com>
Date: Wed, 4 Sep 2019 15:25:12 +0300
Subject: add tests for activity_pub/utils.ex

---
 lib/pleroma/user.ex                          |   1 +
 lib/pleroma/web/activity_pub/activity_pub.ex |  12 +-
 lib/pleroma/web/activity_pub/utils.ex        | 300 +++++++++++++--------------
 3 files changed, 144 insertions(+), 169 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 29fd6d2ea..424ed772f 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -147,6 +147,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,
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index eeb826814..39b46a595 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -435,6 +435,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])
@@ -463,10 +464,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
@@ -478,14 +480,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/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index c9c0c3763..cf82d1a9b 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
 
@@ -188,54 +178,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 =
-      unless fake do
-        %{data: %{"id" => context}, id: context_id} = create_context(map["context"])
+  @spec lazy_put_activity_defaults(map(), boolean) :: map()
+  def lazy_put_activity_defaults(map, fake \\ false)
 
-        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
-
-    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.
   """
@@ -356,23 +351,30 @@ 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]
+    query =
+      from(activity in Activity,
+        where: fragment("data->>'type' = 'Follow'"),
+        where: fragment("data->>'state' = 'pending'"),
+        where: fragment("data->>'actor' = ?", ^actor),
+        where: fragment("data->>'object' = ?", ^object),
+        update: [
+          set: [
+            data: fragment("jsonb_set(data, '{state}', ?)", ^state)
+          ]
+        ]
       )
 
-      User.set_follow_state_cache(actor, object, state)
-      activity = Activity.get_by_id(activity.id)
+    with {_, _} <- Repo.update_all(query, []),
+         {_, _} <- User.set_follow_state_cache(actor, object, state),
+         %Activity{} = activity <- Activity.get_by_id(activity.id) do
       {:ok, activity}
-    rescue
-      e ->
-        {:error, e}
+    else
+      e -> {:error, e}
     end
   end
 
@@ -380,9 +382,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         %Activity{data: %{"actor" => actor, "object" => object}} = activity,
         state
       ) do
-    with new_data <-
-           activity.data
-           |> Map.put("state", state),
+    with new_data <- Map.put(activity.data, "state", state),
          changeset <- Changeset.change(activity, data: new_data),
          {:ok, activity} <- Repo.update(changeset),
          _ <- User.set_follow_state_cache(actor, object, state) do
@@ -411,27 +411,17 @@ defmodule Pleroma.Web.ActivityPub.Utils do
 
   def fetch_latest_follow(%User{ap_id: follower_id}, %User{ap_id: followed_id}) do
     query =
-      from(
-        activity in Activity,
-        where:
-          fragment(
-            "? ->> 'type' = 'Follow'",
-            activity.data
-          ),
-        where: activity.actor == ^follower_id,
-        # this is to use the index
-        where:
-          fragment(
-            "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-            activity.data,
-            activity.data,
-            ^followed_id
-          ),
-        order_by: [fragment("? desc nulls last", activity.id)],
-        limit: 1
-      )
-
-    Repo.one(query)
+      follower_id
+      |> Activity.Queries.by_actor()
+      |> Activity.Queries.by_type("Follow")
+      |> Activity.Queries.by_object_id(followed_id)
+      |> Activity.Queries.limit(1)
+
+    from(
+      activity in query,
+      order_by: [fragment("? desc nulls last", activity.id)]
+    )
+    |> Repo.one()
   end
 
   #### Announce-related helpers
@@ -439,23 +429,14 @@ 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" => id}}) do
-    query =
-      from(
-        activity in Activity,
-        where: activity.actor == ^actor,
-        # this is to use the index
-        where:
-          fragment(
-            "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-            activity.data,
-            activity.data,
-            ^id
-          ),
-        where: fragment("(?)->>'type' = 'Announce'", activity.data)
-      )
-
-    Repo.one(query)
+    actor
+    |> Activity.Queries.by_actor()
+    |> Activity.Queries.by_type("Announce")
+    |> Activity.Queries.by_object_id(id)
+    |> Activity.Queries.limit(1)
+    |> Repo.one()
   end
 
   @doc """
@@ -531,31 +512,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: object.data["announcements"], else: []
+    announcements = fetch_announcements(object)
 
-    with announcements <- [actor | announcements] |> Enum.uniq() do
+    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(fetch_announcements(object), actor) do
       update_element_in_object("announcement", announcements, object)
     end
   end
 
+  defp fetch_announcements(%{data: %{"announcements" => announcements}} = _)
+       when is_list(announcements),
+       do: announcements
+
+  defp fetch_announcements(_), do: []
+
   #### Unfollow-related helpers
 
   def make_unfollow_data(follower, followed, follow_activity, activity_id) do
@@ -569,29 +554,20 @@ 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
     query =
-      from(
-        activity in Activity,
-        where:
-          fragment(
-            "? ->> 'type' = 'Block'",
-            activity.data
-          ),
-        where: activity.actor == ^blocker_id,
-        # this is to use the index
-        where:
-          fragment(
-            "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-            activity.data,
-            activity.data,
-            ^blocked_id
-          ),
-        order_by: [fragment("? desc nulls last", activity.id)],
-        limit: 1
-      )
-
-    Repo.one(query)
+      blocker_id
+      |> Activity.Queries.by_actor()
+      |> Activity.Queries.by_type("Block")
+      |> Activity.Queries.by_object_id(blocked_id)
+      |> Activity.Queries.limit(1)
+
+    from(
+      activity in query,
+      order_by: [fragment("? desc nulls last", activity.id)]
+    )
+    |> Repo.one()
   end
 
   def make_block_data(blocker, blocked, activity_id) do
@@ -631,28 +607,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.
-- 
cgit v1.2.3


From a890451187f0b1507be96ccf144b18fdb8294dd8 Mon Sep 17 00:00:00 2001
From: Maksim Pechnikov <parallel588@gmail.com>
Date: Wed, 4 Sep 2019 17:42:27 +0300
Subject: fetch_announcements -> take_announcements

---
 lib/pleroma/web/activity_pub/utils.ex | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index cf82d1a9b..0d87b9220 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -518,7 +518,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         %Activity{data: %{"actor" => actor, "cc" => [Pleroma.Constants.as_public()]}},
         object
       ) do
-    announcements = fetch_announcements(object)
+    announcements = take_announcements(object)
 
     with announcements <- Enum.uniq([actor | announcements]) do
       update_element_in_object("announcement", announcements, object)
@@ -530,16 +530,16 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   @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
-    with announcements <- List.delete(fetch_announcements(object), actor) do
+    with announcements <- List.delete(take_announcements(object), actor) do
       update_element_in_object("announcement", announcements, object)
     end
   end
 
-  defp fetch_announcements(%{data: %{"announcements" => announcements}} = _)
+  defp take_announcements(%{data: %{"announcements" => announcements}} = _)
        when is_list(announcements),
        do: announcements
 
-  defp fetch_announcements(_), do: []
+  defp take_announcements(_), do: []
 
   #### Unfollow-related helpers
 
-- 
cgit v1.2.3


From 8306078de1abade082f932cda5b8d9297bdcdc80 Mon Sep 17 00:00:00 2001
From: Maksim <parallel588@gmail.com>
Date: Wed, 4 Sep 2019 17:31:14 +0000
Subject: Apply suggestion to lib/pleroma/web/activity_pub/utils.ex

---
 lib/pleroma/web/activity_pub/utils.ex | 33 +++++++++++++--------------------
 1 file changed, 13 insertions(+), 20 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index 0d87b9220..2de02f607 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -356,26 +356,19 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         %Activity{data: %{"actor" => actor, "object" => object}} = activity,
         state
       ) do
-    query =
-      from(activity in Activity,
-        where: fragment("data->>'type' = 'Follow'"),
-        where: fragment("data->>'state' = 'pending'"),
-        where: fragment("data->>'actor' = ?", ^actor),
-        where: fragment("data->>'object' = ?", ^object),
-        update: [
-          set: [
-            data: fragment("jsonb_set(data, '{state}', ?)", ^state)
-          ]
-        ]
-      )
-
-    with {_, _} <- Repo.update_all(query, []),
-         {_, _} <- User.set_follow_state_cache(actor, object, state),
-         %Activity{} = activity <- Activity.get_by_id(activity.id) do
-      {:ok, activity}
-    else
-      e -> {:error, e}
-    end
+   "Follow"
+   |> Activity.Queries.by_type()
+   |> Activity.Queries.by_actor(actor)
+   |> Activity.Queries.by_object_id(object["id"])
+   |> 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}
   end
 
   def update_follow_state(
-- 
cgit v1.2.3


From e2011a667cdf5e67f257c9c30a02c206fb4df913 Mon Sep 17 00:00:00 2001
From: Maksim <parallel588@gmail.com>
Date: Wed, 4 Sep 2019 18:35:01 +0000
Subject: Apply suggestion to lib/pleroma/web/activity_pub/utils.ex

---
 lib/pleroma/web/activity_pub/utils.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index 2de02f607..011acd48e 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -359,7 +359,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
    "Follow"
    |> Activity.Queries.by_type()
    |> Activity.Queries.by_actor(actor)
-   |> Activity.Queries.by_object_id(object["id"])
+   |> Activity.Queries.by_object_id(object)
    |> where(fragment("data->>'state' = 'pending'"))
    |> update(set: [data: fragment("jsonb_set(data, '{state}', ?)", ^state)])
    |> Repo.update_all([])
-- 
cgit v1.2.3


From ae506ca997619f118d18703a9b0802246eb427d5 Mon Sep 17 00:00:00 2001
From: Maksim Pechnikov <parallel588@gmail.com>
Date: Wed, 4 Sep 2019 21:40:53 +0300
Subject: fix formatting

---
 lib/pleroma/web/activity_pub/utils.ex | 26 +++++++++++++-------------
 1 file changed, 13 insertions(+), 13 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index 011acd48e..72e07b59d 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -356,19 +356,19 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         %Activity{data: %{"actor" => actor, "object" => object}} = activity,
         state
       ) do
-   "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}
+    "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}
   end
 
   def update_follow_state(
-- 
cgit v1.2.3


From 769fb778d41df77c2514b5e3c663f3f624c0a266 Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Thu, 12 Sep 2019 21:37:36 +0300
Subject: Track object/create activity fetches

---
 lib/pleroma/delivery.ex                            | 58 ++++++++++++++++++++++
 lib/pleroma/plugs/cache.ex                         | 16 +++++-
 lib/pleroma/user.ex                                | 10 ++++
 .../web/activity_pub/activity_pub_controller.ex    | 29 ++++++++++-
 4 files changed, 111 insertions(+), 2 deletions(-)
 create mode 100644 lib/pleroma/delivery.ex

(limited to 'lib')

diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex
new file mode 100644
index 000000000..f9a9e35cd
--- /dev/null
+++ b/lib/pleroma/delivery.ex
@@ -0,0 +1,58 @@
+# 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.User
+  alias Pleroma.Repo
+  alias Pleroma.Object
+  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])
+    |> 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()
+  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
+
+  def get_or_create(object_id, user_id) do
+    case get(object_id, user_id) do
+      %Delivery{} = delivery -> {:ok, delivery}
+      nil -> create(object_id, user_id)
+    end
+  end
+
+  def delete_all_by_object_id(object_id) do
+    from(d in Delivery, where: d.object_id == ^object_id)
+    |> Repo.delete_all()
+  end
+
+  def get_all_by_object_id(object_id) do
+    from(d in Delivery, where: d.object_id == ^object_id)
+    |> Repo.all()
+  end
+end
diff --git a/lib/pleroma/plugs/cache.ex b/lib/pleroma/plugs/cache.ex
index a81a861d0..42d77fc1f 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,10 @@ 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)
 
@@ -90,7 +95,16 @@ defmodule Pleroma.Plugs.Cache do
         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, {content_type, body, tracking_fun_data}, record, ttl: ttl)
+
+            opts.tracking_fun.(conn, tracking_fun_data)
+          end
 
         put_resp_header(conn, "x-cache", "MISS from Pleroma")
 
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 3aa245f2a..9614acdab 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
@@ -61,6 +62,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()
@@ -1624,4 +1626,12 @@ defmodule Pleroma.User do
   def is_internal_user?(%User{nickname: nil}), do: true
   def is_internal_user?(%User{local: true, nickname: "internal." <> _}), do: true
   def is_internal_user?(_), do: false
+
+  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
 end
diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
index 705dbc1c2..009260d3f 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,15 @@ 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: &Pleroma.Web.ActivityPub.ActivityPubController.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 +63,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 +74,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     end
   end
 
+  def track_object_fetch(conn, object_id) do
+    case conn.assigns[:user] do
+      %User{id: user_id} -> Delivery.create(object_id, user_id)
+      _ -> nil
+    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 +118,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 +129,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: assign(conn, :tracking_fun_data, nil)
+
   defp set_cache_ttl_for(conn, %Activity{object: object}) do
     set_cache_ttl_for(conn, object)
   end
-- 
cgit v1.2.3


From dabc4a00f5cf08dac75f701457a24fce8735175f Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Thu, 12 Sep 2019 22:10:15 +0300
Subject: Put the cache with the right key when using a tracking function

---
 lib/pleroma/plugs/cache.ex | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/plugs/cache.ex b/lib/pleroma/plugs/cache.ex
index 42d77fc1f..50b534e7b 100644
--- a/lib/pleroma/plugs/cache.ex
+++ b/lib/pleroma/plugs/cache.ex
@@ -58,7 +58,8 @@ defmodule Pleroma.Plugs.Cache do
         cache_resp(conn, opts)
 
       {:ok, {content_type, body, tracking_fun_data}} ->
-        conn = opts.tracking_fun(conn, tracking_fun_data)
+        conn = opts.tracking_fun.(conn, tracking_fun_data)
+
         send_cached(conn, {content_type, body})
 
       {:ok, record} ->
@@ -93,7 +94,6 @@ 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}
 
         conn =
           unless opts[:tracking_fun] do
@@ -101,7 +101,7 @@ defmodule Pleroma.Plugs.Cache do
             conn
           else
             tracking_fun_data = Map.get(conn.assigns, :tracking_fun_data, nil)
-            Cachex.put(:web_resp_cache, {content_type, body, tracking_fun_data}, record, ttl: ttl)
+            Cachex.put(:web_resp_cache, key, {content_type, body, tracking_fun_data}, ttl: ttl)
 
             opts.tracking_fun.(conn, tracking_fun_data)
           end
-- 
cgit v1.2.3


From b0e60580215e26caae6452099fa1fbace525937c Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Thu, 12 Sep 2019 22:40:53 +0300
Subject: Parse http signature for request to objects/activities

---
 lib/pleroma/plugs/http_signature.ex | 3 ++-
 lib/pleroma/web/router.ex           | 2 ++
 2 files changed, 4 insertions(+), 1 deletion(-)

(limited to 'lib')

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/web/router.ex b/lib/pleroma/web/router.ex
index 7cd59acb2..badc7e048 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
@@ -513,6 +514,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)
-- 
cgit v1.2.3


From ce23529d917c1830b270a29e774e4ed7768bfeff Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 11:36:49 +0300
Subject: Use delivery info when federating deletes

---
 lib/pleroma/delivery.ex                   |  4 ++++
 lib/pleroma/user.ex                       |  4 ++++
 lib/pleroma/web/activity_pub/publisher.ex | 15 ++++++++++++++-
 3 files changed, 22 insertions(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex
index f9a9e35cd..2e7c019fa 100644
--- a/lib/pleroma/delivery.ex
+++ b/lib/pleroma/delivery.ex
@@ -46,6 +46,10 @@ defmodule Pleroma.Delivery do
     end
   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()
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 9614acdab..785b22643 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -1627,6 +1627,10 @@ 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),
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index c97405690..db64fd2f6 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -5,9 +5,11 @@
 defmodule Pleroma.Web.ActivityPub.Publisher do
   alias Pleroma.Activity
   alias Pleroma.Config
+  alias Pleroma.Delivery
   alias Pleroma.HTTP
   alias Pleroma.Instances
   alias Pleroma.User
+  alias Pleroma.Object
   alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.Transmogrifier
 
@@ -107,7 +109,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
-- 
cgit v1.2.3


From fb96facc32fb275efffeefa2892a1098ecd68b77 Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 12:06:31 +0300
Subject: Remove unused functions and fix credo issues

---
 lib/pleroma/delivery.ex                   | 16 ++--------------
 lib/pleroma/web/activity_pub/publisher.ex |  2 +-
 2 files changed, 3 insertions(+), 15 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex
index 2e7c019fa..ce8fb96f4 100644
--- a/lib/pleroma/delivery.ex
+++ b/lib/pleroma/delivery.ex
@@ -7,9 +7,9 @@ defmodule Pleroma.Delivery do
 
   alias Pleroma.Delivery
   alias Pleroma.FlakeId
-  alias Pleroma.User
-  alias Pleroma.Repo
   alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.User
   alias Pleroma.User
 
   import Ecto.Changeset
@@ -39,13 +39,6 @@ defmodule Pleroma.Delivery do
     |> Repo.one()
   end
 
-  def get_or_create(object_id, user_id) do
-    case get(object_id, user_id) do
-      %Delivery{} = delivery -> {:ok, delivery}
-      nil -> create(object_id, user_id)
-    end
-  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, []}
@@ -54,9 +47,4 @@ defmodule Pleroma.Delivery do
     from(d in Delivery, where: d.object_id == ^object_id)
     |> Repo.delete_all()
   end
-
-  def get_all_by_object_id(object_id) do
-    from(d in Delivery, where: d.object_id == ^object_id)
-    |> Repo.all()
-  end
 end
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index db64fd2f6..c39e89a6a 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -8,8 +8,8 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
   alias Pleroma.Delivery
   alias Pleroma.HTTP
   alias Pleroma.Instances
-  alias Pleroma.User
   alias Pleroma.Object
+  alias Pleroma.User
   alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.Transmogrifier
 
-- 
cgit v1.2.3


From 517017048316a52172d60d26b03beddb85af7b39 Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 10:09:46 +0000
Subject: Apply suggestion to
 lib/pleroma/web/activity_pub/activity_pub_controller.ex

---
 lib/pleroma/web/activity_pub/activity_pub_controller.ex | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
index 009260d3f..025641722 100644
--- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
@@ -26,10 +26,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
 
   plug(
     Pleroma.Plugs.Cache,
-    [
-      query_params: false,
-      tracking_fun: &Pleroma.Web.ActivityPub.ActivityPubController.track_object_fetch/2
-    ]
+    [query_params: false, tracking_fun: &__MODULE__.track_object_fetch/2]
     when action in [:activity, :object]
   )
 
-- 
cgit v1.2.3


From 3896a51b8aefe6fe54251ffd559c636980faa87e Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 10:09:56 +0000
Subject: Apply suggestion to
 lib/pleroma/web/activity_pub/activity_pub_controller.ex

---
 lib/pleroma/web/activity_pub/activity_pub_controller.ex | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
index 025641722..4bd13defb 100644
--- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
@@ -72,9 +72,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
   end
 
   def track_object_fetch(conn, object_id) do
-    case conn.assigns[:user] do
-      %User{id: user_id} -> Delivery.create(object_id, user_id)
-      _ -> nil
+    with %{assigns: %{user: %User{id: user_id}}} <- conn do
+      Delivery.create(object_id, user_id)
     end
 
     conn
-- 
cgit v1.2.3


From 2784962dba295ee35677e93996df53d1711e5768 Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 15:23:03 +0000
Subject: Apply suggestion to
 lib/pleroma/web/activity_pub/activity_pub_controller.ex

---
 lib/pleroma/web/activity_pub/activity_pub_controller.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
index 4bd13defb..70d4a5baf 100644
--- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
@@ -130,7 +130,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     assign(conn, :tracking_fun_data, object_id)
   end
 
-  defp maybe_set_tracking_data(conn, _activity), do: assign(conn, :tracking_fun_data, nil)
+  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)
-- 
cgit v1.2.3


From 05f8a066a107af2f7151aee8d85af97cf6a4835c Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 15:23:26 +0000
Subject: Apply suggestion to lib/pleroma/delivery.ex

---
 lib/pleroma/delivery.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex
index ce8fb96f4..38c148c34 100644
--- a/lib/pleroma/delivery.ex
+++ b/lib/pleroma/delivery.ex
@@ -31,7 +31,7 @@ defmodule Pleroma.Delivery do
   def create(object_id, user_id) do
     %Delivery{}
     |> changeset(%{user_id: user_id, object_id: object_id})
-    |> Repo.insert()
+    |> Repo.insert(on_conflict: :nothing)
   end
 
   def get(object_id, user_id) do
-- 
cgit v1.2.3


From 8900cb68aef535dbf60de87fce47d85b91909077 Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 15:25:15 +0000
Subject: Apply suggestion to
 lib/pleroma/web/activity_pub/activity_pub_controller.ex

---
 lib/pleroma/web/activity_pub/activity_pub_controller.ex | 2 ++
 1 file changed, 2 insertions(+)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
index 70d4a5baf..01b34fb1d 100644
--- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex
@@ -71,6 +71,8 @@ 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)
-- 
cgit v1.2.3


From ac4a748fad34c02647bf72e802cd9d74205681fe Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Sep 2019 19:28:35 +0300
Subject: Disallow NULLs in deliveries

---
 lib/pleroma/delivery.ex | 1 +
 1 file changed, 1 insertion(+)

(limited to 'lib')

diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex
index 38c148c34..29a1e5a77 100644
--- a/lib/pleroma/delivery.ex
+++ b/lib/pleroma/delivery.ex
@@ -23,6 +23,7 @@ defmodule Pleroma.Delivery do
   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)
-- 
cgit v1.2.3


From b4cf74c1067b866574a63fbd25ccb12cc1fed619 Mon Sep 17 00:00:00 2001
From: Maksim Pechnikov <parallel588@gmail.com>
Date: Sun, 15 Sep 2019 14:53:58 +0300
Subject: added prepare html for RichMedia.Parser

---
 lib/pleroma/web/rich_media/parser.ex | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

(limited to 'lib')

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
 
-- 
cgit v1.2.3


From aab264db82054df470075c65ca25c42bbcc5d7a8 Mon Sep 17 00:00:00 2001
From: Steven Fuchs <steven.fuchs@dockyard.com>
Date: Mon, 16 Sep 2019 07:44:03 +0000
Subject: Streamer refactoring

---
 lib/pleroma/activity/ir/topics.ex                 |  63 +++++
 lib/pleroma/application.ex                        |   2 +-
 lib/pleroma/notification.ex                       |   6 +-
 lib/pleroma/web/activity_pub/activity_pub.ex      |  50 +---
 lib/pleroma/web/mastodon_api/websocket_handler.ex |   7 +-
 lib/pleroma/web/streamer.ex                       | 318 ----------------------
 lib/pleroma/web/streamer/ping.ex                  |  33 +++
 lib/pleroma/web/streamer/state.ex                 |  68 +++++
 lib/pleroma/web/streamer/streamer.ex              |  55 ++++
 lib/pleroma/web/streamer/streamer_socket.ex       |  31 +++
 lib/pleroma/web/streamer/supervisor.ex            |  33 +++
 lib/pleroma/web/streamer/worker.ex                | 220 +++++++++++++++
 lib/pleroma/web/views/streamer_view.ex            |  66 +++++
 13 files changed, 590 insertions(+), 362 deletions(-)
 create mode 100644 lib/pleroma/activity/ir/topics.ex
 delete mode 100644 lib/pleroma/web/streamer.ex
 create mode 100644 lib/pleroma/web/streamer/ping.ex
 create mode 100644 lib/pleroma/web/streamer/state.ex
 create mode 100644 lib/pleroma/web/streamer/streamer.ex
 create mode 100644 lib/pleroma/web/streamer/streamer_socket.ex
 create mode 100644 lib/pleroma/web/streamer/supervisor.ex
 create mode 100644 lib/pleroma/web/streamer/worker.ex
 create mode 100644 lib/pleroma/web/views/streamer_view.ex

(limited to 'lib')

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..3b37ce630 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -141,7 +141,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),
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/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 41f6a0f1f..bc5ae7fbf 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
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/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..7b5199068
--- /dev/null
+++ b/lib/pleroma/web/streamer/state.ex
@@ -0,0 +1,68 @@
+defmodule Pleroma.Web.Streamer.State do
+  use GenServer
+  require Logger
+
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
+  end
+
+  def add_socket(topic, socket) do
+    GenServer.call(__MODULE__, {:add, socket, topic})
+  end
+
+  def remove_socket(topic, socket) do
+    GenServer.call(__MODULE__, {:remove, socket, topic})
+  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, socket, topic}, _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, socket, topic}, _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 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
-- 
cgit v1.2.3


From c623b4324deaf236334a0f77a81435b5bffadf3c Mon Sep 17 00:00:00 2001
From: kaniini <ariadne@dereferenced.org>
Date: Mon, 16 Sep 2019 09:09:21 +0000
Subject: Revert "Merge branch 'streamer-refactoring' into 'develop'"

This reverts merge request !1653
---
 lib/pleroma/activity/ir/topics.ex                 |  63 -----
 lib/pleroma/application.ex                        |   2 +-
 lib/pleroma/notification.ex                       |   6 +-
 lib/pleroma/web/activity_pub/activity_pub.ex      |  50 +++-
 lib/pleroma/web/mastodon_api/websocket_handler.ex |   7 +-
 lib/pleroma/web/streamer.ex                       | 318 ++++++++++++++++++++++
 lib/pleroma/web/streamer/ping.ex                  |  33 ---
 lib/pleroma/web/streamer/state.ex                 |  68 -----
 lib/pleroma/web/streamer/streamer.ex              |  55 ----
 lib/pleroma/web/streamer/streamer_socket.ex       |  31 ---
 lib/pleroma/web/streamer/supervisor.ex            |  33 ---
 lib/pleroma/web/streamer/worker.ex                | 220 ---------------
 lib/pleroma/web/views/streamer_view.ex            |  66 -----
 13 files changed, 362 insertions(+), 590 deletions(-)
 delete mode 100644 lib/pleroma/activity/ir/topics.ex
 create mode 100644 lib/pleroma/web/streamer.ex
 delete mode 100644 lib/pleroma/web/streamer/ping.ex
 delete mode 100644 lib/pleroma/web/streamer/state.ex
 delete mode 100644 lib/pleroma/web/streamer/streamer.ex
 delete mode 100644 lib/pleroma/web/streamer/streamer_socket.ex
 delete mode 100644 lib/pleroma/web/streamer/supervisor.ex
 delete mode 100644 lib/pleroma/web/streamer/worker.ex
 delete mode 100644 lib/pleroma/web/views/streamer_view.ex

(limited to 'lib')

diff --git a/lib/pleroma/activity/ir/topics.ex b/lib/pleroma/activity/ir/topics.ex
deleted file mode 100644
index 010897abc..000000000
--- a/lib/pleroma/activity/ir/topics.ex
+++ /dev/null
@@ -1,63 +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.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 3b37ce630..49094704b 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -141,7 +141,7 @@ defmodule Pleroma.Application do
   defp streamer_child(:test), do: []
 
   defp streamer_child(_) do
-    [Pleroma.Web.Streamer.supervisor()]
+    [Pleroma.Web.Streamer]
   end
 
   defp oauth_cleanup_child(true),
diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex
index 8012389ac..b7c880c51 100644
--- a/lib/pleroma/notification.ex
+++ b/lib/pleroma/notification.ex
@@ -210,10 +210,8 @@ defmodule Pleroma.Notification do
     unless skip?(activity, user) do
       notification = %Notification{user_id: user.id, activity: activity}
       {:ok, notification} = Repo.insert(notification)
-
-      ["user", "user:notification"]
-      |> Streamer.stream(notification)
-
+      Streamer.stream("user", notification)
+      Streamer.stream("user:notification", notification)
       Push.send(notification)
       notification
     end
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index bc5ae7fbf..41f6a0f1f 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -4,7 +4,6 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Activity
-  alias Pleroma.Activity.Ir.Topics
   alias Pleroma.Config
   alias Pleroma.Conversation
   alias Pleroma.Notification
@@ -17,7 +16,6 @@ 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
 
@@ -189,7 +187,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       participations
       |> Repo.preload(:user)
 
-    Streamer.stream("participation", participations)
+    Enum.each(participations, fn participation ->
+      Pleroma.Web.Streamer.stream("participation", participation)
+    end)
   end
 
   def stream_out_participations(%Object{data: %{"context" => context}}, user) do
@@ -208,15 +208,41 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   def stream_out_participations(_, _), do: :noop
 
-  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
+  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
   end
 
   def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
index 3c26eb406..dbd3542ea 100644
--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
@@ -8,7 +8,6 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   alias Pleroma.Repo
   alias Pleroma.User
   alias Pleroma.Web.OAuth.Token
-  alias Pleroma.Web.Streamer
 
   @behaviour :cowboy_websocket
 
@@ -25,7 +24,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   ]
   @anonymous_streams ["public", "public:local", "hashtag"]
 
-  # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
+  # Handled by periodic keepalive in Pleroma.Web.Streamer.
   @timeout :infinity
 
   def init(%{qs: qs} = req, state) do
@@ -66,7 +65,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
       }, topic #{state.topic}"
     )
 
-    Streamer.add_socket(state.topic, streamer_socket(state))
+    Pleroma.Web.Streamer.add_socket(state.topic, streamer_socket(state))
     {:ok, state}
   end
 
@@ -81,7 +80,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
       }, topic #{state.topic || "?"}: #{inspect(reason)}"
     )
 
-    Streamer.remove_socket(state.topic, streamer_socket(state))
+    Pleroma.Web.Streamer.remove_socket(state.topic, streamer_socket(state))
     :ok
   end
 
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
new file mode 100644
index 000000000..587c43f40
--- /dev/null
+++ b/lib/pleroma/web/streamer.ex
@@ -0,0 +1,318 @@
+# 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
deleted file mode 100644
index f77cbb95c..000000000
--- a/lib/pleroma/web/streamer/ping.ex
+++ /dev/null
@@ -1,33 +0,0 @@
-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
deleted file mode 100644
index 7b5199068..000000000
--- a/lib/pleroma/web/streamer/state.ex
+++ /dev/null
@@ -1,68 +0,0 @@
-defmodule Pleroma.Web.Streamer.State do
-  use GenServer
-  require Logger
-
-  alias Pleroma.Web.Streamer.StreamerSocket
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
-  end
-
-  def add_socket(topic, socket) do
-    GenServer.call(__MODULE__, {:add, socket, topic})
-  end
-
-  def remove_socket(topic, socket) do
-    GenServer.call(__MODULE__, {:remove, socket, topic})
-  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, socket, topic}, _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, socket, topic}, _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 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
deleted file mode 100644
index 8cf719277..000000000
--- a/lib/pleroma/web/streamer/streamer.ex
+++ /dev/null
@@ -1,55 +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
-  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
deleted file mode 100644
index f006c0306..000000000
--- a/lib/pleroma/web/streamer/streamer_socket.ex
+++ /dev/null
@@ -1,31 +0,0 @@
-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
deleted file mode 100644
index 6afe19323..000000000
--- a/lib/pleroma/web/streamer/supervisor.ex
+++ /dev/null
@@ -1,33 +0,0 @@
-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
deleted file mode 100644
index 5804508eb..000000000
--- a/lib/pleroma/web/streamer/worker.ex
+++ /dev/null
@@ -1,220 +0,0 @@
-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
deleted file mode 100644
index b13030fa0..000000000
--- a/lib/pleroma/web/views/streamer_view.ex
+++ /dev/null
@@ -1,66 +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.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
-- 
cgit v1.2.3


From 96816ceaa25c21cec7677e75dcddd7ffb42d83c3 Mon Sep 17 00:00:00 2001
From: Egor Kislitsyn <egor@kislitsyn.com>
Date: Mon, 16 Sep 2019 17:03:37 +0700
Subject: Revert "Merge branch 'revert-4fabf83a' into 'develop'"

This reverts commit fe7fd331263007e0fb2877ef7370a09a9704da36, reversing
changes made to 4fabf83ad01352442906d79187aeab4c777f4df8.
---
 lib/pleroma/activity/ir/topics.ex                 |  63 +++++
 lib/pleroma/application.ex                        |   2 +-
 lib/pleroma/notification.ex                       |   6 +-
 lib/pleroma/web/activity_pub/activity_pub.ex      |  50 +---
 lib/pleroma/web/mastodon_api/websocket_handler.ex |   7 +-
 lib/pleroma/web/streamer.ex                       | 318 ----------------------
 lib/pleroma/web/streamer/ping.ex                  |  33 +++
 lib/pleroma/web/streamer/state.ex                 |  68 +++++
 lib/pleroma/web/streamer/streamer.ex              |  55 ++++
 lib/pleroma/web/streamer/streamer_socket.ex       |  31 +++
 lib/pleroma/web/streamer/supervisor.ex            |  33 +++
 lib/pleroma/web/streamer/worker.ex                | 220 +++++++++++++++
 lib/pleroma/web/views/streamer_view.ex            |  66 +++++
 13 files changed, 590 insertions(+), 362 deletions(-)
 create mode 100644 lib/pleroma/activity/ir/topics.ex
 delete mode 100644 lib/pleroma/web/streamer.ex
 create mode 100644 lib/pleroma/web/streamer/ping.ex
 create mode 100644 lib/pleroma/web/streamer/state.ex
 create mode 100644 lib/pleroma/web/streamer/streamer.ex
 create mode 100644 lib/pleroma/web/streamer/streamer_socket.ex
 create mode 100644 lib/pleroma/web/streamer/supervisor.ex
 create mode 100644 lib/pleroma/web/streamer/worker.ex
 create mode 100644 lib/pleroma/web/views/streamer_view.ex

(limited to 'lib')

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..3b37ce630 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -141,7 +141,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),
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/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 41f6a0f1f..bc5ae7fbf 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
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/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..7b5199068
--- /dev/null
+++ b/lib/pleroma/web/streamer/state.ex
@@ -0,0 +1,68 @@
+defmodule Pleroma.Web.Streamer.State do
+  use GenServer
+  require Logger
+
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
+  end
+
+  def add_socket(topic, socket) do
+    GenServer.call(__MODULE__, {:add, socket, topic})
+  end
+
+  def remove_socket(topic, socket) do
+    GenServer.call(__MODULE__, {:remove, socket, topic})
+  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, socket, topic}, _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, socket, topic}, _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 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
-- 
cgit v1.2.3


From 085d014f0859b3b3e5023c423ae0361ec6ed6c67 Mon Sep 17 00:00:00 2001
From: Egor Kislitsyn <egor@kislitsyn.com>
Date: Mon, 16 Sep 2019 19:26:00 +0700
Subject: Fix `Transmogrifier.upgrade_user_from_ap_id/1`

---
 lib/pleroma/web/activity_pub/transmogrifier.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index acb3087d0..8461b666e 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
@@ -1050,7 +1050,7 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
     with %User{local: false} = user <- User.get_cached_by_ap_id(ap_id),
          {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id),
          already_ap <- User.ap_enabled?(user),
-         {:ok, user} <- user |> User.upgrade_changeset(data) |> User.update_and_set_cache() do
+         {:ok, user} <- user |> User.upgrade_changeset(data, true) |> User.update_and_set_cache() do
       unless already_ap do
         TransmogrifierWorker.enqueue("user_upgrade", %{"user_id" => user.id})
       end
-- 
cgit v1.2.3


From a21584556f2c3edb90db3c58ba2a4829a7e220c1 Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Tue, 17 Sep 2019 13:04:43 +0300
Subject: Update oban to 0.8.1

This version uses a different locking mechanism, which gets rid of
`WARNING:  you don't own a lock of type ShareLock` log spam
---
 lib/pleroma/flake_id.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

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__
-- 
cgit v1.2.3


From 80c5c3495bdd7939e576c8746a959f3f89f44042 Mon Sep 17 00:00:00 2001
From: Steven Fuchs <steven.fuchs@dockyard.com>
Date: Tue, 17 Sep 2019 14:44:52 +0000
Subject: remove remaining errors from tests

---
 lib/pleroma/application.ex        | 53 +++++++++++++++++++++++++++------------
 lib/pleroma/web/streamer/state.ex | 18 ++++++++++---
 2 files changed, 51 insertions(+), 20 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 3b37ce630..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?()) ++
@@ -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/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex
index 7b5199068..c48752d95 100644
--- a/lib/pleroma/web/streamer/state.ex
+++ b/lib/pleroma/web/streamer/state.ex
@@ -4,16 +4,18 @@ defmodule Pleroma.Web.Streamer.State do
 
   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, socket, topic})
+    GenServer.call(__MODULE__, {:add, topic, socket})
   end
 
   def remove_socket(topic, socket) do
-    GenServer.call(__MODULE__, {:remove, socket, topic})
+    do_remove_socket(@env, topic, socket)
   end
 
   def get_sockets do
@@ -29,7 +31,7 @@ defmodule Pleroma.Web.Streamer.State do
     {:reply, state, state}
   end
 
-  def handle_call({:add, socket, topic}, _from, %{sockets: sockets} = state) do
+  def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
     internal_topic = internal_topic(topic, socket)
     stream_socket = StreamerSocket.from_socket(socket)
 
@@ -44,7 +46,7 @@ defmodule Pleroma.Web.Streamer.State do
     {:reply, state, state}
   end
 
-  def handle_call({:remove, socket, topic}, _from, %{sockets: sockets} = state) do
+  def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
     internal_topic = internal_topic(topic, socket)
     stream_socket = StreamerSocket.from_socket(socket)
 
@@ -57,6 +59,14 @@ defmodule Pleroma.Web.Streamer.State do
     {: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}"
-- 
cgit v1.2.3


From 6193157f1998b10ac6cb9f4d36dd863eced81b37 Mon Sep 17 00:00:00 2001
From: Steven Fuchs <steven.fuchs@dockyard.com>
Date: Tue, 17 Sep 2019 18:12:27 +0000
Subject: Fix notification warnings

---
 lib/pleroma/workers/web_pusher_worker.ex | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

(limited to 'lib')

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
-- 
cgit v1.2.3