Merge branch 'fix/streaming-with-irreversible-filter' into 'develop'

Irreversible filter works with user streaming.

See merge request pleroma/pleroma!3542

changelog.d/3542.fix

@@ -0,0 +1 @@
+Irreversible filter works with user streaming.

lib/pleroma/filter.ex

@@ -48,6 +48,14 @@ defmodule Pleroma.Filter do
     from(f in query, where: f.hide)
   end
 
+  @spec get_context(Ecto.Query.t(), String.t()) :: Ecto.Query.t()
+  def get_context(query, context) do
+    from(
+      f in query,
+      where: ^context in f.context
+    )
+  end
+
   @spec get_filters(Ecto.Query.t() | module(), User.t()) :: [t()]
   def get_filters(query \\ __MODULE__, %User{id: user_id}) do
     query =
@@ -196,18 +204,22 @@ defmodule Pleroma.Filter do
     end
   end
 
-  @spec compose_regex(User.t() | [t()], format()) :: String.t() | Regex.t() | nil
-  def compose_regex(user_or_filters, format \\ :postgres)
+  @spec compose_regex(User.t(), String.t(), format()) :: String.t() | Regex.t() | nil
+  def compose_regex(user, context, format \\ :postgres)
 
-  def compose_regex(%User{} = user, format) do
+  def compose_regex(%User{} = user, context, format)
+      when context in ["home", "notifications"] do
     __MODULE__
     |> get_active()
     |> get_irreversible()
+    |> get_context(context)
     |> get_filters(user)
-    |> compose_regex(format)
+    |> to_regex(format)
   end
 
-  def compose_regex([_ | _] = filters, format) do
+  def compose_regex(_, _, _), do: nil
+
+  defp to_regex([_ | _] = filters, format) do
     phrases =
       filters
       |> Enum.map(& &1.phrase)
@@ -222,5 +234,5 @@ defmodule Pleroma.Filter do
     end
   end
 
-  def compose_regex(_, _), do: nil
+  defp to_regex(_, _), do: nil
 end

lib/pleroma/notification.ex

@@ -171,7 +171,7 @@ defmodule Pleroma.Notification do
   end
 
   defp exclude_filtered(query, user) do
-    case Pleroma.Filter.compose_regex(user) do
+    case Pleroma.Filter.compose_regex(user, "notifications") do
       nil ->
         query
 
@@ -722,7 +722,7 @@ defmodule Pleroma.Notification do
       object.data["actor"] == user.ap_id ->
         false
 
-      not is_nil(regex = Pleroma.Filter.compose_regex(user, :re)) ->
+      not is_nil(regex = Pleroma.Filter.compose_regex(user, "notifications", :re)) ->
         Regex.match?(regex, object.data["content"])
 
       true ->

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1241,8 +1241,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp restrict_instance(query, _), do: query
 
-  defp restrict_filtered(query, %{user: %User{} = user}) do
-    case Filter.compose_regex(user) do
+  defp restrict_filtered(query, %{phrase_filtering_user: %User{} = user}) do
+    case Filter.compose_regex(user, "home") do
       nil ->
         query
 
@@ -1255,10 +1255,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
-  defp restrict_filtered(query, %{blocking_user: %User{} = user}) do
-    restrict_filtered(query, %{user: user})
-  end
-
   defp restrict_filtered(query, _), do: query
 
   defp restrict_unauthenticated(query, nil) do

lib/pleroma/web/mastodon_api/controllers/timeline_controller.ex

@@ -52,6 +52,7 @@ defmodule Pleroma.Web.MastodonAPI.TimelineController do
       |> Map.put(:muting_user, user)
       |> Map.put(:reply_filtering_user, user)
       |> Map.put(:announce_filtering_user, user)
+      |> Map.put(:phrase_filtering_user, user)
       |> Map.put(:user, user)
       |> Map.put(:local_only, params[:local])
       |> Map.put(:followed_hashtags, followed_hashtags)

lib/pleroma/web/mastodon_api/websocket_handler.ex

@@ -85,7 +85,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   def handle_info({:render_with_user, view, template, item, topic}, state) do
     user = %User{} = User.get_cached_by_ap_id(state.user.ap_id)
 
-    unless Streamer.filtered_by_user?(user, item) do
+    unless Streamer.filtered_by_user?(topic, user, item) do
       message = view.render(template, item, user, topic)
       {:push, {:text, message}, %{state | user: user}}
     else

lib/pleroma/web/streamer.ex

@@ -184,9 +184,15 @@ defmodule Pleroma.Web.Streamer do
     end
   end
 
-  def filtered_by_user?(user, item, streamed_type \\ :activity)
+  @spec filtered_by_user?(
+          topic :: String.t(),
+          User.t(),
+          Activity.t() | Notification.t(),
+          :activity | :notification
+        ) :: boolean()
+  def filtered_by_user?(topic, user, item, streamed_type \\ :activity)
 
-  def filtered_by_user?(%User{} = user, %Activity{} = item, streamed_type) do
+  def filtered_by_user?(topic, %User{} = user, %Activity{} = item, streamed_type) do
     %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
       User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])
 
@@ -202,6 +208,9 @@ defmodule Pleroma.Web.Streamer do
                parent.data["actor"] == user.ap_id),
          true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
          true <- MapSet.disjoint?(recipients, recipient_blocks),
+         false <-
+           streamed_type == :activity && topic == "user:" <> to_string(user.id) &&
+             match_irreversible_filter(user, parent),
          %{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),
@@ -214,8 +223,8 @@ defmodule Pleroma.Web.Streamer do
     end
   end
 
-  def filtered_by_user?(%User{} = user, %Notification{activity: activity}, _) do
-    filtered_by_user?(user, activity, :notification)
+  def filtered_by_user?(topic, %User{} = user, %Notification{activity: activity}, _) do
+    filtered_by_user?(topic, user, activity, :notification)
   end
 
   defp do_stream("direct", item) do
@@ -412,6 +421,18 @@ defmodule Pleroma.Web.Streamer do
     end
   end
 
+  defp match_irreversible_filter(user, %Object{data: %{"content" => content}}) do
+    case Pleroma.Filter.compose_regex(user, "home", :re) do
+      nil ->
+        false
+
+      regex ->
+        Regex.match?(regex, content)
+    end
+  end
+
+  defp match_irreversible_filter(_, _), do: false
+
   # In dev/prod the streamer registry is expected to be started, so return true
   # In test it is possible to have the registry started for a test so it will check
   # In benchmark it will never find the process alive and return false

test/pleroma/notification_test.exs

@@ -268,7 +268,7 @@ defmodule Pleroma.NotificationTest do
       subscriber = insert(:user)
 
       User.subscribe(subscriber, user)
-      insert(:filter, user: subscriber, phrase: "cofe", hide: true)
+      insert(:filter, user: subscriber, phrase: "cofe", context: ["notifications"], hide: true)
 
       {:ok, status} = CommonAPI.post(user, %{status: "got cofe?"})
 
@@ -1068,7 +1068,7 @@ defmodule Pleroma.NotificationTest do
     end
 
     test "it doesn't return notifications about mentions with filtered word", %{user: user} do
-      insert(:filter, user: user, phrase: "cofe", hide: true)
+      insert(:filter, user: user, phrase: "cofe", context: ["notifications"], hide: true)
       another_user = insert(:user)
 
       {:ok, _activity} = CommonAPI.post(another_user, %{status: "@#{user.nickname} got cofe?"})

test/pleroma/web/activity_pub/activity_pub_test.exs

@@ -912,7 +912,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
     test "doesn't return activities with filtered words" do
       user = insert(:user)
       user_two = insert(:user)
-      insert(:filter, user: user, phrase: "test", hide: true)
+      insert(:filter, user: user, phrase: "test", context: ["home"], hide: true)
 
       {:ok, %{id: id1, data: %{"context" => context}}} = CommonAPI.post(user, %{status: "1"})
 
@@ -928,7 +928,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
 
       activities =
         context
-        |> ActivityPub.fetch_activities_for_context(%{user: user})
+        |> ActivityPub.fetch_activities_for_context(%{user: user, phrase_filtering_user: user})
         |> Enum.map(& &1.id)
 
       assert length(activities) == 4
@@ -1275,13 +1275,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       user = insert(:user)
       user_two = insert(:user)
 
-      insert(:filter, user: user_two, phrase: "cofe", hide: true)
-      insert(:filter, user: user_two, phrase: "ok boomer", hide: true)
-      insert(:filter, user: user_two, phrase: "test", hide: false)
+      insert(:filter, user: user_two, phrase: "cofe", context: ["home", "public"], hide: true)
+      insert(:filter, user: user_two, phrase: "ok boomer", context: ["home"], hide: true)
+      insert(:filter, user: user_two, phrase: "noise", context: ["notification"], hide: true)
+      insert(:filter, user: user_two, phrase: "test", context: ["home"], hide: false)
 
       params = %{
         type: ["Create", "Announce"],
-        user: user_two
+        user: user_two,
+        phrase_filtering_user: user_two
       }
 
       {:ok, %{user: user, user_two: user_two, params: params}}
@@ -1324,6 +1326,32 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       assert Enum.empty?(activities)
     end
 
+    test "it does not filter if filter is hide or doesn't have home in context", %{
+      user: user,
+      params: params
+    } do
+      {:ok, _} = CommonAPI.post(user, %{status: "make some noise"})
+      {:ok, _} = CommonAPI.post(user, %{status: "test"})
+
+      activities = ActivityPub.fetch_activities([], params)
+
+      assert Enum.count(activities) == 2
+    end
+
+    test "it returns all statuses in public timeline", %{user: user, user_two: user_two} do
+      {:ok, _} = CommonAPI.post(user, %{status: "got cofe?"})
+      {:ok, _} = CommonAPI.post(user, %{status: "test!"})
+
+      activities =
+        ActivityPub.fetch_public_activities(%{
+          type: ["Create"],
+          muting_user: user_two,
+          blocking_user: user_two
+        })
+
+      assert Enum.count(activities) == 2
+    end
+
     test "it returns all statuses if user does not have any filters" do
       another_user = insert(:user)
       {:ok, _} = CommonAPI.post(another_user, %{status: "got cofe?"})
@@ -1332,7 +1360,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       activities =
         ActivityPub.fetch_activities([], %{
           type: ["Create", "Announce"],
-          user: another_user
+          user: another_user,
+          phrase_filtering_user: another_user
         })
 
       assert Enum.count(activities) == 2

test/pleroma/web/streamer_test.exs

@@ -243,42 +243,42 @@ defmodule Pleroma.Web.StreamerTest do
     end
 
     test "it streams the user's post in the 'user' stream", %{user: user, token: oauth_token} do
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
 
       assert_receive {:render_with_user, _, _, ^activity, _}
-      refute Streamer.filtered_by_user?(user, activity)
+      refute Streamer.filtered_by_user?(topic, user, activity)
     end
 
     test "it streams boosts of the user in the 'user' stream", %{user: user, token: oauth_token} do
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
 
       other_user = insert(:user)
       {:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
       {:ok, announce} = CommonAPI.repeat(activity.id, user)
 
       assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce, _}
-      refute Streamer.filtered_by_user?(user, announce)
+      refute Streamer.filtered_by_user?(topic, user, announce)
     end
 
     test "it does not stream announces of the user's own posts in the 'user' stream", %{
       user: user,
       token: oauth_token
     } do
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
 
       other_user = insert(:user)
       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
       {:ok, announce} = CommonAPI.repeat(activity.id, other_user)
 
-      assert Streamer.filtered_by_user?(user, announce)
+      assert Streamer.filtered_by_user?(topic, user, announce)
     end
 
     test "it does stream notifications announces of the user's own posts in the 'user' stream", %{
       user: user,
       token: oauth_token
     } do
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
 
       other_user = insert(:user)
       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
@@ -289,14 +289,14 @@ defmodule Pleroma.Web.StreamerTest do
         |> Repo.get_by(%{user_id: user.id, activity_id: announce.id})
         |> Repo.preload(:activity)
 
-      refute Streamer.filtered_by_user?(user, notification)
+      refute Streamer.filtered_by_user?(topic, user, notification)
     end
 
     test "it streams boosts of mastodon user in the 'user' stream", %{
       user: user,
       token: oauth_token
     } do
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
 
       other_user = insert(:user)
       {:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
@@ -311,7 +311,7 @@ defmodule Pleroma.Web.StreamerTest do
         Pleroma.Web.ActivityPub.Transmogrifier.handle_incoming(data)
 
       assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce, _}
-      refute Streamer.filtered_by_user?(user, announce)
+      refute Streamer.filtered_by_user?(topic, user, announce)
     end
 
     test "it sends notify to in the 'user' stream", %{
@@ -319,11 +319,11 @@ defmodule Pleroma.Web.StreamerTest do
       token: oauth_token,
       notify: notify
     } do
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
       Streamer.stream("user", notify)
 
       assert_receive {:render_with_user, _, _, ^notify, _}
-      refute Streamer.filtered_by_user?(user, notify)
+      refute Streamer.filtered_by_user?(topic, user, notify)
     end
 
     test "it sends notify to in the 'user:notification' stream", %{
@@ -331,11 +331,11 @@ defmodule Pleroma.Web.StreamerTest do
       token: oauth_token,
       notify: notify
     } do
-      Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
       Streamer.stream("user:notification", notify)
 
       assert_receive {:render_with_user, _, _, ^notify, _}
-      refute Streamer.filtered_by_user?(user, notify)
+      refute Streamer.filtered_by_user?(topic, user, notify)
     end
 
     test "it sends chat messages to the 'user:pleroma_chat' stream", %{
@@ -401,11 +401,11 @@ defmodule Pleroma.Web.StreamerTest do
         Repo.get_by(Pleroma.Notification, user_id: user.id, activity_id: create_activity.id)
         |> Repo.preload(:activity)
 
-      Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
       Streamer.stream("user:notification", notify)
 
       assert_receive {:render_with_user, _, _, ^notify, _}
-      refute Streamer.filtered_by_user?(user, notify)
+      refute Streamer.filtered_by_user?(topic, user, notify)
     end
 
     test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
@@ -432,12 +432,12 @@ defmodule Pleroma.Web.StreamerTest do
       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
       {:ok, _} = CommonAPI.add_mute(activity, user)
 
-      Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 
       {:ok, favorite_activity} = CommonAPI.favorite(activity.id, user2)
 
       refute_receive _
-      assert Streamer.filtered_by_user?(user, favorite_activity)
+      assert Streamer.filtered_by_user?(topic, user, favorite_activity)
     end
 
     test "it sends favorite to 'user:notification' stream'", %{
@@ -447,12 +447,12 @@ defmodule Pleroma.Web.StreamerTest do
       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 
       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
-      Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
       {:ok, favorite_activity} = CommonAPI.favorite(activity.id, user2)
 
       assert_receive {:render_with_user, _, "notification.json", notif, _}
       assert notif.activity.id == favorite_activity.id
-      refute Streamer.filtered_by_user?(user, notif)
+      refute Streamer.filtered_by_user?(topic, user, notif)
     end
 
     test "it doesn't send the 'user:notification' stream' when a domain is blocked", %{
@@ -463,11 +463,11 @@ defmodule Pleroma.Web.StreamerTest do
 
       {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
-      Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
       {:ok, favorite_activity} = CommonAPI.favorite(activity.id, user2)
 
       refute_receive _
-      assert Streamer.filtered_by_user?(user, favorite_activity)
+      assert Streamer.filtered_by_user?(topic, user, favorite_activity)
     end
 
     test "it sends follow activities to the 'user:notification' stream", %{
@@ -476,12 +476,12 @@ defmodule Pleroma.Web.StreamerTest do
     } do
       user2 = insert(:user)
 
-      Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
       {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user, user2)
 
       assert_receive {:render_with_user, _, "notification.json", notif, _}
       assert notif.activity.id == follow_activity.id
-      refute Streamer.filtered_by_user?(user, notif)
+      refute Streamer.filtered_by_user?(topic, user, notif)
     end
 
     test "it sends follow relationships updates to the 'user' stream", %{
@@ -540,23 +540,23 @@ defmodule Pleroma.Web.StreamerTest do
 
       {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
 
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
       {:ok, edited} = CommonAPI.update(activity, sender, %{status: "mew mew"})
       create = Pleroma.Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
 
       assert_receive {:render_with_user, _, "status_update.json", ^create, _}
-      refute Streamer.filtered_by_user?(user, edited)
+      refute Streamer.filtered_by_user?(topic, user, edited)
     end
 
     test "it streams own edits in the 'user' stream", %{user: user, token: oauth_token} do
       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
 
-      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
       {:ok, edited} = CommonAPI.update(activity, user, %{status: "mew mew"})
       create = Pleroma.Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
 
       assert_receive {:render_with_user, _, "status_update.json", ^create, _}
-      refute Streamer.filtered_by_user?(user, edited)
+      refute Streamer.filtered_by_user?(topic, user, edited)
     end
 
     test "it streams posts containing followed hashtags on the 'user' stream", %{
@@ -595,11 +595,11 @@ defmodule Pleroma.Web.StreamerTest do
       %{user: user, token: oauth_token} = oauth_access(["read"])
       other_user = insert(:user)
 
-      Streamer.get_topic_and_add_socket("public", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", user, oauth_token)
 
       {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
       assert_receive {:render_with_user, _, _, ^activity, _}
-      refute Streamer.filtered_by_user?(other_user, activity)
+      refute Streamer.filtered_by_user?(topic, other_user, activity)
     end
 
     test "it sends to public (unauthenticated)" do
@@ -634,7 +634,7 @@ defmodule Pleroma.Web.StreamerTest do
     test "it streams edits in the 'public' stream" do
       sender = insert(:user)
 
-      Streamer.get_topic_and_add_socket("public", nil, nil)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", nil, nil)
       {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
       assert_receive {:text, _}
 
@@ -647,13 +647,13 @@ defmodule Pleroma.Web.StreamerTest do
       assert_receive {:text, event}
       assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
       assert %{"id" => ^activity_id} = Jason.decode!(payload)
-      refute Streamer.filtered_by_user?(sender, edited)
+      refute Streamer.filtered_by_user?(topic, sender, edited)
     end
 
     test "it streams multiple edits in the 'public' stream correctly" do
       sender = insert(:user)
 
-      Streamer.get_topic_and_add_socket("public", nil, nil)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", nil, nil)
       {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
       assert_receive {:text, _}
 
@@ -666,7 +666,7 @@ defmodule Pleroma.Web.StreamerTest do
       assert_receive {:text, event}
       assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
       assert %{"id" => ^activity_id} = Jason.decode!(payload)
-      refute Streamer.filtered_by_user?(sender, edited)
+      refute Streamer.filtered_by_user?(topic, sender, edited)
 
       {:ok, edited} = CommonAPI.update(activity, sender, %{status: "mew mew 2"})
 
@@ -676,7 +676,7 @@ defmodule Pleroma.Web.StreamerTest do
       assert_receive {:text, event}
       assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
       assert %{"id" => ^activity_id, "content" => "mew mew 2"} = Jason.decode!(payload)
-      refute Streamer.filtered_by_user?(sender, edited)
+      refute Streamer.filtered_by_user?(topic, sender, edited)
     end
   end
 
@@ -696,10 +696,10 @@ defmodule Pleroma.Web.StreamerTest do
             )
         )
 
-      Streamer.get_topic_and_add_socket("public", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", user, oauth_token)
       Streamer.stream("public", activity)
       assert_receive {:render_with_user, _, _, ^activity, _}
-      assert Streamer.filtered_by_user?(user, activity)
+      assert Streamer.filtered_by_user?(topic, user, activity)
     end
 
     test "it sends message if recipients invalid and thread containment is disabled" do
@@ -717,11 +717,11 @@ defmodule Pleroma.Web.StreamerTest do
             )
         )
 
-      Streamer.get_topic_and_add_socket("public", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", user, oauth_token)
       Streamer.stream("public", activity)
 
       assert_receive {:render_with_user, _, _, ^activity, _}
-      refute Streamer.filtered_by_user?(user, activity)
+      refute Streamer.filtered_by_user?(topic, user, activity)
     end
 
     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
@@ -740,11 +740,11 @@ defmodule Pleroma.Web.StreamerTest do
             )
         )
 
-      Streamer.get_topic_and_add_socket("public", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", user, oauth_token)
       Streamer.stream("public", activity)
 
       assert_receive {:render_with_user, _, _, ^activity, _}
-      refute Streamer.filtered_by_user?(user, activity)
+      refute Streamer.filtered_by_user?(topic, user, activity)
     end
   end
 
@@ -755,10 +755,10 @@ defmodule Pleroma.Web.StreamerTest do
       blocked_user = insert(:user)
       {:ok, _user_relationship} = User.block(user, blocked_user)
 
-      Streamer.get_topic_and_add_socket("public", user, oauth_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", user, oauth_token)
       {:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
       assert_receive {:render_with_user, _, _, ^activity, _}
-      assert Streamer.filtered_by_user?(user, activity)
+      assert Streamer.filtered_by_user?(topic, user, activity)
     end
 
     test "it filters messages transitively involving blocked users", %{
@@ -768,24 +768,24 @@ defmodule Pleroma.Web.StreamerTest do
       blockee = insert(:user)
       friend = insert(:user)
 
-      Streamer.get_topic_and_add_socket("public", blocker, blocker_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", blocker, blocker_token)
 
       {:ok, _user_relationship} = User.block(blocker, blockee)
 
       {:ok, activity_one} = CommonAPI.post(friend, %{status: "hey! @#{blockee.nickname}"})
 
       assert_receive {:render_with_user, _, _, ^activity_one, _}
-      assert Streamer.filtered_by_user?(blocker, activity_one)
+      assert Streamer.filtered_by_user?(topic, blocker, activity_one)
 
       {:ok, activity_two} = CommonAPI.post(blockee, %{status: "hey! @#{friend.nickname}"})
 
       assert_receive {:render_with_user, _, _, ^activity_two, _}
-      assert Streamer.filtered_by_user?(blocker, activity_two)
+      assert Streamer.filtered_by_user?(topic, blocker, activity_two)
 
       {:ok, activity_three} = CommonAPI.post(blockee, %{status: "hey! @#{blocker.nickname}"})
 
       assert_receive {:render_with_user, _, _, ^activity_three, _}
-      assert Streamer.filtered_by_user?(blocker, activity_three)
+      assert Streamer.filtered_by_user?(topic, blocker, activity_three)
     end
   end
 
@@ -837,7 +837,8 @@ defmodule Pleroma.Web.StreamerTest do
       {:ok, list} = List.create("Test", user_a)
       {:ok, list} = List.follow(list, user_b)
 
-      Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
+      {:ok, topic} =
+        Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 
       {:ok, activity} =
         CommonAPI.post(user_b, %{
@@ -846,7 +847,7 @@ defmodule Pleroma.Web.StreamerTest do
         })
 
       assert_receive {:render_with_user, _, _, ^activity, _}
-      refute Streamer.filtered_by_user?(user_a, activity)
+      refute Streamer.filtered_by_user?(topic, user_a, activity)
     end
   end
 
@@ -861,10 +862,10 @@ defmodule Pleroma.Web.StreamerTest do
 
       {:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})
 
-      Streamer.get_topic_and_add_socket("user", user1, user1_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user1, user1_token)
       {:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
       assert_receive {:render_with_user, _, _, ^announce_activity, _}
-      assert Streamer.filtered_by_user?(user1, announce_activity)
+      assert Streamer.filtered_by_user?(topic, user1, announce_activity)
     end
 
     test "it filters reblog notification for reblog-muted actors", %{
@@ -876,11 +877,11 @@ defmodule Pleroma.Web.StreamerTest do
       CommonAPI.hide_reblogs(user2, user1)
 
       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
-      Streamer.get_topic_and_add_socket("user", user1, user1_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user1, user1_token)
       {:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 
       assert_receive {:render_with_user, _, "notification.json", notif, _}
-      assert Streamer.filtered_by_user?(user1, notif)
+      assert Streamer.filtered_by_user?(topic, user1, notif)
     end
 
     test "it send non-reblog notification for reblog-muted actors", %{
@@ -892,11 +893,11 @@ defmodule Pleroma.Web.StreamerTest do
       CommonAPI.hide_reblogs(user2, user1)
 
       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
-      Streamer.get_topic_and_add_socket("user", user1, user1_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user1, user1_token)
       {:ok, _favorite_activity} = CommonAPI.favorite(create_activity.id, user2)
 
       assert_receive {:render_with_user, _, "notification.json", notif, _}
-      refute Streamer.filtered_by_user?(user1, notif)
+      refute Streamer.filtered_by_user?(topic, user1, notif)
     end
   end
 
@@ -904,14 +905,68 @@ defmodule Pleroma.Web.StreamerTest do
     test "it filters posts from muted threads" do
       user = insert(:user)
       %{user: user2, token: user2_token} = oauth_access(["read"])
-      Streamer.get_topic_and_add_socket("user", user2, user2_token)
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user2, user2_token)
 
       {:ok, user2, user, _activity} = CommonAPI.follow(user, user2)
       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
       {:ok, _} = CommonAPI.add_mute(activity, user2)
 
       assert_receive {:render_with_user, _, _, ^activity, _}
-      assert Streamer.filtered_by_user?(user2, activity)
+      assert Streamer.filtered_by_user?(topic, user2, activity)
+    end
+  end
+
+  describe "word filters" do
+    setup do
+      %{user: user, token: token} = oauth_access(["read"])
+      user_two = insert(:user)
+      {:ok, user, user_two, _activity} = CommonAPI.follow(user, user_two)
+      insert(:filter, user: user, phrase: "cofe", context: ["home", "public"], hide: true)
+      insert(:filter, user: user, phrase: "ok boomer", context: ["home"], hide: true)
+      insert(:filter, user: user, phrase: "test", context: ["home"], hide: false)
+      {:ok, %{user: user, token: token, user_two: user_two}}
+    end
+
+    test "it filters posts from hard word filters", %{
+      user: user,
+      user_two: user_two,
+      token: oauth_token
+    } do
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
+
+      {:ok, activity_one} = CommonAPI.post(user_two, %{status: "Give me some cofe!"})
+
+      assert_receive {:render_with_user, _, _, ^activity_one, ^topic}
+      assert Streamer.filtered_by_user?(topic, user, activity_one)
+
+      {:ok, activity_two} = CommonAPI.post(user_two, %{status: "ok boomer"})
+
+      assert_receive {:render_with_user, _, _, ^activity_two, ^topic}
+      assert Streamer.filtered_by_user?(topic, user, activity_two)
+    end
+
+    test "it doesn't filters posts from soft word filters", %{
+      user: user,
+      user_two: user_two,
+      token: oauth_token
+    } do
+      {:ok, topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      {:ok, activity} = CommonAPI.post(user_two, %{status: "test"})
+
+      assert_receive {:render_with_user, _, _, ^activity, ^topic}
+      refute Streamer.filtered_by_user?(topic, user, activity)
+    end
+
+    test "it doesn't filters posts from public word filters", %{
+      user: user,
+      user_two: user_two,
+      token: oauth_token
+    } do
+      {:ok, topic} = Streamer.get_topic_and_add_socket("public", user, oauth_token)
+      {:ok, activity} = CommonAPI.post(user_two, %{status: "Give me some cofe!"})
+
+      assert_receive {:render_with_user, _, _, ^activity, ^topic}
+      refute Streamer.filtered_by_user?(topic, user, activity)
     end
   end