Scope the unread rooms stream per user

UnreadRoomsChannel streamed from a hardcoded global name, and every message
published its room id to it. Any authenticated user, including one with no
memberships at all, could subscribe and watch which rooms were active and exactly
when, across every closed room and direct conversation on the account. The browser
filters ids it doesn't recognise, but that happens after delivery.

Its sibling ReadRoomsChannel is already scoped per user; this mirrors it, and
message creation fans the notice out to the room's members instead of broadcasting
it to everyone. No message content was exposed either way, only the timing.

test/channels/unread_rooms_channel_test.rb

@@ -0,0 +1,47 @@
+require "test_helper"
+
+class UnreadRoomsChannelTest < ActionCable::Channel::TestCase
+  test "streams only the subscriber's own unread stream" do
+    stub_connection(current_user: users(:jz))
+
+    subscribe
+
+    assert subscription.confirmed?
+    assert_has_stream "user_#{users(:jz).id}_unreads"
+    assert_not_includes subscription.streams, "unread_rooms"
+  end
+
+  test "an outsider is not told about activity in a room they can't see" do
+    direct = rooms(:bender_and_kevin)
+    assert_not direct.users.include?(users(:jz)), "jz must be an outsider for this test to mean anything"
+
+    broadcasts = capture_unread_broadcasts_for(users(:jz)) do
+      direct.messages.create!(body: "Private", creator: users(:kevin), client_message_id: "outsider").broadcast_create
+    end
+
+    assert_empty broadcasts
+  end
+
+  test "a member is told about activity in their own room" do
+    direct = rooms(:bender_and_kevin)
+
+    broadcasts = capture_unread_broadcasts_for(users(:kevin)) do
+      direct.messages.create!(body: "Private", creator: users(:bender), client_message_id: "member").broadcast_create
+    end
+
+    assert_equal [ direct.id ], broadcasts.collect { |broadcast| broadcast["roomId"] }
+  end
+
+  private
+    def capture_unread_broadcasts_for(user)
+      stub_connection(current_user: user)
+      subscribe
+
+      stream = subscription.streams.sole
+      before = ActionCable.server.pubsub.broadcasts(stream).size
+
+      yield
+
+      ActionCable.server.pubsub.broadcasts(stream).drop(before).collect { |broadcast| JSON.parse(broadcast) }
+    end
+end

test/controllers/messages_controller_test.rb

@@ -57,9 +57,22 @@ class MessagesControllerTest < ActionDispatch::IntegrationTest
     end
   end
 
-  test "creating a message broadcasts unread room" do
-    assert_broadcasts "unread_rooms", 1 do
-      post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one", client_message_id: 999 } }
+  test "creating a message broadcasts unread room to each member" do
+    @room.users.each do |member|
+      assert_broadcasts UnreadRoomsChannel.stream_name_for(member.id), 1 do
+        post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one #{member.id}", client_message_id: member.id } }
+      end
+    end
+  end
+
+  test "creating a message doesn't broadcast unread room to non-members" do
+    outsiders = User.where.not(id: @room.users.map(&:id))
+    assert outsiders.any?, "need someone outside the room for this test to mean anything"
+
+    outsiders.each do |outsider|
+      assert_no_broadcasts UnreadRoomsChannel.stream_name_for(outsider.id) do
+        post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one", client_message_id: 999 } }
+      end
     end
   end