add pub/sub broadcast model, replace iterate-and-filter (Phase A, Step 5)

Introduce topic-based PubSub for O(subscribers) message routing:
- Clients subscribe to "global" and "user:{id}" on connect
- Channel broadcasts route through "channel:{id}" topics
- deliverBroadcast() uses PubSub instead of iterating all clients
- UnsubscribeAll on disconnect/kick cleans up subscriptions
- Sequence numbering and replay buffer preserved unchanged

https://claude.ai/code/session_01CBFF3r84ywkJRWwuqw8zD8

Server/ws/hub.go

@@ -41,6 +41,8 @@ type Hub struct {
 	registry     *HandlerRegistry
 	permChecker  *permissions.Checker
 
+	pubsub *PubSub // topic-based pub/sub for O(subscribers) broadcast
+
 	seq            uint64           // atomic monotonic sequence counter
 	seqMu          syncutil.Mutex   // serializes seq assignment + replay insertion + delivery order
 	replayBuf      *EventRingBuffer // recent broadcast events for reconnection replay
@@ -74,6 +76,7 @@ func NewHub(database *db.DB, limiter *auth.RateLimiter, svc *service.Services) *
 		register:        make(chan *Client, 32),
 		unregister:      make(chan *Client, 32),
 		stop:            make(chan struct{}),
+		pubsub:          NewPubSub(),
 		replayBuf:       NewEventRingBuffer(1000),
 		registry:        reg,
 		permChecker:     permissions.NewChecker(database),
@@ -386,6 +389,9 @@ func (h *Hub) registerNow(c *Client) {
 		// by the handshake path in serve.go, which runs before registerNow.
 		// registerNow only handles in-memory client replacement.
 
+		// Remove the old client from all pub/sub topics before replacing.
+		h.pubsub.UnsubscribeAll(old)
+
 		// Kick the stale connection atomically before registering
 		// the new one — prevents TOCTOU races on duplicate login.
 		slog.Warn("hub: kicking stale connection for re-registering user",
@@ -395,17 +401,23 @@ func (h *Hub) registerNow(c *Client) {
 	h.clients[c.userID] = c
 	slog.Info("hub: client registered", "user_id", c.userID, "total_clients", len(h.clients))
 	h.mu.Unlock()
+
+	// Subscribe the new client to default pub/sub topics.
+	h.pubsub.Subscribe(c, TopicGlobal)
+	h.pubsub.Subscribe(c, UserTopic(c.userID))
 }
 
 func (h *Hub) unregisterNow(c *Client) bool {
 	h.mu.Lock()
-	defer h.mu.Unlock()
 	current, exists := h.clients[c.userID]
 	if exists && current == c {
 		delete(h.clients, c.userID)
 		slog.Info("hub: client unregistered", "user_id", c.userID, "total_clients", len(h.clients))
+		h.mu.Unlock()
+		h.pubsub.UnsubscribeAll(c)
 		return false // not replaced
 	}
+	h.mu.Unlock()
 	return true // different client registered = was replaced
 }
 
@@ -553,6 +565,7 @@ func (h *Hub) kickClient(c *Client) {
 		delete(h.clients, c.userID)
 	}
 	h.mu.Unlock()
+	h.pubsub.UnsubscribeAll(c)
 	c.closeSend()
 }
 
@@ -700,7 +713,7 @@ func (h *Hub) sweepStaleVoiceStates() {
 }
 
 // deliverBroadcast stamps bm.msg with a monotonic sequence number, stores it
-// in the replay buffer, and sends it to the appropriate clients.
+// in the replay buffer, and sends it to the appropriate clients via pub/sub.
 func (h *Hub) deliverBroadcast(bm broadcastMsg) {
 	h.seqMu.Lock()
 	defer h.seqMu.Unlock()
@@ -711,27 +724,14 @@ func (h *Hub) deliverBroadcast(bm broadcastMsg) {
 	// Store in replay buffer for reconnection recovery.
 	h.replayBuf.Push(seq, bm.channelID, msg)
 
-	h.mu.RLock()
-	defer h.mu.RUnlock()
-
-	delivered := 0
-	skipped := 0
-	for _, c := range h.clients {
-		// channelID == 0 → global broadcast, deliver to everyone.
-		// Otherwise, only deliver to clients viewing this channel
-		// (via channel_focus) or in voice on this channel.
-		// Clients that haven't focused any channel (channelID == 0)
-		// are intentionally excluded from channel-scoped broadcasts
-		// to prevent information leakage (BUG-122).
-		if bm.channelID != 0 && c.getChannelID() != bm.channelID && c.getVoiceChID() != bm.channelID {
-			skipped++
-			continue
-		}
-		c.sendMsg(msg)
-		delivered++
-	}
-	if bm.channelID != 0 {
+	if bm.channelID == 0 {
+		// Global broadcast — deliver to every connected client.
+		h.pubsub.PublishGlobal(msg)
+	} else {
+		// Channel-scoped broadcast — deliver to subscribers of the channel topic.
+		topic := ChannelTopic(bm.channelID)
+		delivered := h.pubsub.Publish(topic, msg, 0)
 		slog.Debug("hub: channel broadcast",
-			"channel_id", bm.channelID, "delivered", delivered, "skipped", skipped, "seq", seq)
+			"channel_id", bm.channelID, "delivered", delivered, "seq", seq)
 	}
 }

Server/ws/pubsub.go

@@ -0,0 +1,192 @@
+package ws
+
+import (
+	"fmt"
+	"log/slog"
+	"sync"
+)
+
+// Topic is a named pub/sub channel that clients can subscribe to.
+// Naming convention:
+//
+//	"global"       – all clients, subscribed on connect
+//	"channel:42"   – text channel messages
+//	"voice:7"      – voice channel events
+//	"user:123"     – per-user direct events (DMs, mentions)
+type Topic string
+
+// TopicGlobal is the well-known topic every client subscribes to on connect.
+const TopicGlobal Topic = "global"
+
+// ChannelTopic returns the topic for a text channel.
+func ChannelTopic(channelID int64) Topic {
+	return Topic(fmt.Sprintf("channel:%d", channelID))
+}
+
+// VoiceTopic returns the topic for a voice channel.
+func VoiceTopic(channelID int64) Topic {
+	return Topic(fmt.Sprintf("voice:%d", channelID))
+}
+
+// UserTopic returns the per-user topic for DMs and mentions.
+func UserTopic(userID int64) Topic {
+	return Topic(fmt.Sprintf("user:%d", userID))
+}
+
+// PubSub provides topic-based publish/subscribe routing for WebSocket clients.
+// Broadcasting to a topic costs O(subscribers) instead of O(all connections).
+//
+// Thread-safe: all methods may be called from any goroutine.
+type PubSub struct {
+	mu sync.RWMutex
+
+	// Forward index: topic → (userID → *Client)
+	topics map[Topic]map[int64]*Client
+
+	// Reverse index: userID → set of topics (for efficient UnsubscribeAll)
+	clients map[int64]map[Topic]struct{}
+}
+
+// NewPubSub creates an empty PubSub ready for use.
+func NewPubSub() *PubSub {
+	return &PubSub{
+		topics:  make(map[Topic]map[int64]*Client),
+		clients: make(map[int64]map[Topic]struct{}),
+	}
+}
+
+// Subscribe registers client for messages on the given topic.
+// If the client is already subscribed, this is a no-op.
+func (ps *PubSub) Subscribe(client *Client, topic Topic) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+
+	// Forward index
+	subs, ok := ps.topics[topic]
+	if !ok {
+		subs = make(map[int64]*Client)
+		ps.topics[topic] = subs
+	}
+	subs[client.userID] = client
+
+	// Reverse index
+	ts, ok := ps.clients[client.userID]
+	if !ok {
+		ts = make(map[Topic]struct{})
+		ps.clients[client.userID] = ts
+	}
+	ts[topic] = struct{}{}
+}
+
+// Unsubscribe removes client from the given topic.
+// No-op if the client is not subscribed.
+func (ps *PubSub) Unsubscribe(client *Client, topic Topic) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	ps.unsubscribeLocked(client.userID, topic)
+}
+
+// unsubscribeLocked removes userID from topic. Caller must hold ps.mu (write).
+func (ps *PubSub) unsubscribeLocked(userID int64, topic Topic) {
+	// Forward index
+	if subs, ok := ps.topics[topic]; ok {
+		delete(subs, userID)
+		if len(subs) == 0 {
+			delete(ps.topics, topic)
+		}
+	}
+
+	// Reverse index
+	if ts, ok := ps.clients[userID]; ok {
+		delete(ts, topic)
+		if len(ts) == 0 {
+			delete(ps.clients, userID)
+		}
+	}
+}
+
+// UnsubscribeAll removes client from every topic it is subscribed to.
+// Called when a client disconnects.
+func (ps *PubSub) UnsubscribeAll(client *Client) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+
+	ts, ok := ps.clients[client.userID]
+	if !ok {
+		return
+	}
+
+	// Remove from every topic's subscriber set.
+	for topic := range ts {
+		if subs, ok := ps.topics[topic]; ok {
+			delete(subs, client.userID)
+			if len(subs) == 0 {
+				delete(ps.topics, topic)
+			}
+		}
+	}
+
+	// Remove the reverse-index entry entirely.
+	delete(ps.clients, client.userID)
+}
+
+// Publish sends msg to all subscribers of topic, except the client identified
+// by excludeUserID (pass 0 to exclude nobody). Returns the number of clients
+// the message was delivered to.
+func (ps *PubSub) Publish(topic Topic, msg []byte, excludeUserID int64) int {
+	ps.mu.RLock()
+	subs := ps.topics[topic]
+	// Snapshot the subscriber slice under read lock to avoid holding the lock
+	// while calling sendMsg (which acquires the client's own mutex).
+	clients := make([]*Client, 0, len(subs))
+	for uid, c := range subs {
+		if uid != excludeUserID {
+			clients = append(clients, c)
+		}
+	}
+	ps.mu.RUnlock()
+
+	for _, c := range clients {
+		c.sendMsg(msg)
+	}
+	return len(clients)
+}
+
+// PublishGlobal sends msg to every client subscribed to the "global" topic.
+// This is equivalent to Publish(TopicGlobal, msg, 0) but makes intent explicit.
+func (ps *PubSub) PublishGlobal(msg []byte) int {
+	return ps.Publish(TopicGlobal, msg, 0)
+}
+
+// SubscriberCount returns the number of subscribers for a topic.
+func (ps *PubSub) SubscriberCount(topic Topic) int {
+	ps.mu.RLock()
+	defer ps.mu.RUnlock()
+	return len(ps.topics[topic])
+}
+
+// TopicsForClient returns the set of topics a client is subscribed to.
+// Intended for debugging and tests.
+func (ps *PubSub) TopicsForClient(userID int64) []Topic {
+	ps.mu.RLock()
+	defer ps.mu.RUnlock()
+	ts := ps.clients[userID]
+	result := make([]Topic, 0, len(ts))
+	for t := range ts {
+		result = append(result, t)
+	}
+	return result
+}
+
+// debugDump logs the current subscription state. For development use only.
+func (ps *PubSub) debugDump() {
+	ps.mu.RLock()
+	defer ps.mu.RUnlock()
+	for topic, subs := range ps.topics {
+		ids := make([]int64, 0, len(subs))
+		for uid := range subs {
+			ids = append(ids, uid)
+		}
+		slog.Debug("pubsub: topic", "topic", string(topic), "subscribers", ids)
+	}
+}

Server/ws/pubsub_test.go

@@ -0,0 +1,263 @@
+package ws
+
+import (
+	"sort"
+	"sync"
+	"testing"
+	"time"
+)
+
+func newTestPubSub() *PubSub {
+	return NewPubSub()
+}
+
+func makeTestClient(userID int64) *Client {
+	return &Client{
+		userID: userID,
+		send:   make(chan []byte, 16),
+	}
+}
+
+// ─── Subscribe / Unsubscribe ─────────────────────────────────────────────────
+
+func TestPubSub_Subscribe(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(1)
+
+	ps.Subscribe(c, "channel:42")
+
+	if n := ps.SubscriberCount("channel:42"); n != 1 {
+		t.Fatalf("SubscriberCount = %d, want 1", n)
+	}
+}
+
+func TestPubSub_SubscribeIdempotent(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(1)
+
+	ps.Subscribe(c, "channel:42")
+	ps.Subscribe(c, "channel:42") // duplicate
+
+	if n := ps.SubscriberCount("channel:42"); n != 1 {
+		t.Fatalf("SubscriberCount = %d after double subscribe, want 1", n)
+	}
+}
+
+func TestPubSub_Unsubscribe(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(1)
+
+	ps.Subscribe(c, "channel:42")
+	ps.Unsubscribe(c, "channel:42")
+
+	if n := ps.SubscriberCount("channel:42"); n != 0 {
+		t.Fatalf("SubscriberCount = %d after unsubscribe, want 0", n)
+	}
+}
+
+func TestPubSub_UnsubscribeNonExistent(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(1)
+
+	// Should not panic.
+	ps.Unsubscribe(c, "channel:99")
+}
+
+func TestPubSub_UnsubscribeAll(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(1)
+
+	ps.Subscribe(c, "channel:1")
+	ps.Subscribe(c, "channel:2")
+	ps.Subscribe(c, TopicGlobal)
+
+	ps.UnsubscribeAll(c)
+
+	if n := ps.SubscriberCount("channel:1"); n != 0 {
+		t.Fatalf("channel:1 still has %d subscribers", n)
+	}
+	if n := ps.SubscriberCount("channel:2"); n != 0 {
+		t.Fatalf("channel:2 still has %d subscribers", n)
+	}
+	if n := ps.SubscriberCount(TopicGlobal); n != 0 {
+		t.Fatalf("global still has %d subscribers", n)
+	}
+	if topics := ps.TopicsForClient(1); len(topics) != 0 {
+		t.Fatalf("client still has topics: %v", topics)
+	}
+}
+
+func TestPubSub_UnsubscribeAllEmpty(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(99)
+
+	// Should not panic on client with no subscriptions.
+	ps.UnsubscribeAll(c)
+}
+
+// ─── Publish ─────────────────────────────────────────────────────────────────
+
+func TestPubSub_Publish(t *testing.T) {
+	ps := newTestPubSub()
+	c1 := makeTestClient(1)
+	c2 := makeTestClient(2)
+	c3 := makeTestClient(3) // not subscribed
+
+	ps.Subscribe(c1, "channel:42")
+	ps.Subscribe(c2, "channel:42")
+
+	msg := []byte(`{"type":"chat"}`)
+	delivered := ps.Publish("channel:42", msg, 0)
+
+	if delivered != 2 {
+		t.Fatalf("delivered = %d, want 2", delivered)
+	}
+
+	assertChanMsg(t, c1.send, msg)
+	assertChanMsg(t, c2.send, msg)
+	assertChanEmpty(t, c3.send)
+}
+
+func TestPubSub_PublishExclude(t *testing.T) {
+	ps := newTestPubSub()
+	c1 := makeTestClient(1)
+	c2 := makeTestClient(2)
+
+	ps.Subscribe(c1, "channel:42")
+	ps.Subscribe(c2, "channel:42")
+
+	msg := []byte(`{"type":"typing"}`)
+	delivered := ps.Publish("channel:42", msg, 1) // exclude user 1
+
+	if delivered != 1 {
+		t.Fatalf("delivered = %d, want 1", delivered)
+	}
+
+	assertChanEmpty(t, c1.send)
+	assertChanMsg(t, c2.send, msg)
+}
+
+func TestPubSub_PublishEmptyTopic(t *testing.T) {
+	ps := newTestPubSub()
+
+	delivered := ps.Publish("channel:999", []byte(`{}`), 0)
+	if delivered != 0 {
+		t.Fatalf("delivered = %d for empty topic, want 0", delivered)
+	}
+}
+
+func TestPubSub_PublishGlobal(t *testing.T) {
+	ps := newTestPubSub()
+	c1 := makeTestClient(1)
+	c2 := makeTestClient(2)
+
+	ps.Subscribe(c1, TopicGlobal)
+	ps.Subscribe(c2, TopicGlobal)
+
+	msg := []byte(`{"type":"presence"}`)
+	delivered := ps.PublishGlobal(msg)
+
+	if delivered != 2 {
+		t.Fatalf("delivered = %d, want 2", delivered)
+	}
+
+	assertChanMsg(t, c1.send, msg)
+	assertChanMsg(t, c2.send, msg)
+}
+
+// ─── TopicsForClient ─────────────────────────────────────────────────────────
+
+func TestPubSub_TopicsForClient(t *testing.T) {
+	ps := newTestPubSub()
+	c := makeTestClient(1)
+
+	ps.Subscribe(c, TopicGlobal)
+	ps.Subscribe(c, "channel:10")
+	ps.Subscribe(c, UserTopic(1))
+
+	topics := ps.TopicsForClient(1)
+	sort.Slice(topics, func(i, j int) bool { return topics[i] < topics[j] })
+
+	expected := []Topic{"channel:10", TopicGlobal, UserTopic(1)}
+	if len(topics) != len(expected) {
+		t.Fatalf("topics = %v, want %v", topics, expected)
+	}
+	for i := range expected {
+		if topics[i] != expected[i] {
+			t.Fatalf("topics[%d] = %q, want %q", i, topics[i], expected[i])
+		}
+	}
+}
+
+// ─── Topic helpers ───────────────────────────────────────────────────────────
+
+func TestChannelTopic(t *testing.T) {
+	if got := ChannelTopic(42); got != "channel:42" {
+		t.Fatalf("ChannelTopic(42) = %q", got)
+	}
+}
+
+func TestVoiceTopic(t *testing.T) {
+	if got := VoiceTopic(7); got != "voice:7" {
+		t.Fatalf("VoiceTopic(7) = %q", got)
+	}
+}
+
+func TestUserTopic(t *testing.T) {
+	if got := UserTopic(123); got != "user:123" {
+		t.Fatalf("UserTopic(123) = %q", got)
+	}
+}
+
+// ─── Concurrency safety ─────────────────────────────────────────────────────
+
+func TestPubSub_ConcurrentAccess(t *testing.T) {
+	ps := newTestPubSub()
+	const N = 50
+
+	var wg sync.WaitGroup
+	wg.Add(N * 3) // subscribe + publish + unsubscribe
+
+	for i := 0; i < N; i++ {
+		c := makeTestClient(int64(i))
+		go func() {
+			defer wg.Done()
+			ps.Subscribe(c, "channel:1")
+		}()
+		go func() {
+			defer wg.Done()
+			ps.Publish("channel:1", []byte(`{"x":1}`), 0)
+		}()
+		go func(c *Client) {
+			defer wg.Done()
+			ps.UnsubscribeAll(c)
+		}(c)
+	}
+
+	wg.Wait()
+	// No panics or data races = pass. (Run with -race.)
+}
+
+// ─── helpers ─────────────────────────────────────────────────────────────────
+
+func assertChanMsg(t *testing.T, ch <-chan []byte, want []byte) {
+	t.Helper()
+	select {
+	case got := <-ch:
+		if string(got) != string(want) {
+			t.Errorf("got %q, want %q", got, want)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("expected message but channel was empty")
+	}
+}
+
+func assertChanEmpty(t *testing.T, ch <-chan []byte) {
+	t.Helper()
+	select {
+	case msg := <-ch:
+		t.Errorf("expected empty channel but got %q", msg)
+	default:
+		// ok
+	}
+}