fix(ws): revoke channel-topic subscriptions on role change (F1)

READ_MESSAGES was authorized once at channel_focus and then frozen into a
durable pub/sub subscription that no role change re-evaluated, so a demoted
user kept receiving every message posted in channels their new role can no
longer read. BroadcastMemberUpdate now recomputes the allowed set from the
user's current role and unsubscribes each held channel topic it no longer
covers, evicting the socket if visibility cannot be resolved.

Verified by a panel of agents; both added tests were confirmed failing
against the unpatched tree.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Server/ws/hub.go

@@ -730,9 +730,98 @@ func (h *Hub) BroadcastUserUpdate(userID int64, username string, avatar *string,
 	h.BroadcastToAll(buildUserUpdate(userID, username, avatar, identityPublicKey))
 }
 
-// BroadcastMemberUpdate sends a member_update message to all connected clients.
+// BroadcastMemberUpdate sends a member_update message to all connected clients
+// and re-evaluates the reassigned user's live channel subscriptions.
 func (h *Hub) BroadcastMemberUpdate(userID int64, roleName string) {
 	h.BroadcastToAll(buildMemberUpdate(userID, roleName))
+	h.revokeUnreadableChannels(userID)
+}
+
+// revokeUnreadableChannels drops the channel-topic subscriptions the user's new
+// role may no longer READ. READ_MESSAGES is checked once, at channel_focus, and
+// then becomes a durable pub/sub subscription, so without this a demoted user
+// keeps receiving every chat_message / chat_edited / reaction_update posted in
+// the channels their old role could read for as long as the socket stays open.
+//
+// The per-client work mirrors RefreshChannelVisibility, the channel_overrides
+// equivalent: targeted, unsequenced channel_delete + Unsubscribe (a replayed
+// channel_delete would be filtered by the allowed set computed at replay time),
+// then a visibilityChangeSeq bump so a client resuming across this change takes
+// the full-ready path instead of replay.
+//
+// Only the topics the socket actually holds are examined — a blanket sweep over
+// every channel would disclose the full channel-ID list to a demoted user.
+func (h *Hub) revokeUnreadableChannels(userID int64) {
+	// Stored after the targeted sends (as in RefreshChannelVisibility) so a
+	// concurrent seq advance errs toward re-syncing more clients. Deferred
+	// because it must cover the early returns too: a user who is offline, or
+	// whose socket is closed below, converges via the full-ready path.
+	defer h.visibilityChangeSeq.Store(atomic.LoadUint64(&h.seq))
+
+	if h.db == nil {
+		return
+	}
+	h.mu.RLock()
+	c, ok := h.clients[userID]
+	h.mu.RUnlock()
+	if !ok || c.user == nil {
+		return
+	}
+
+	// Called via the admin HubBroadcaster interface, which carries no context;
+	// the re-evaluation must complete regardless of the triggering request.
+	ctx := context.Background()
+
+	// c.user is a connect-time snapshot and the role just changed, so resolve
+	// the current user — and through it the current role — from the DB.
+	var allowed map[int64]bool
+	user, err := h.db.GetUserByID(ctx, userID)
+	if err == nil && user != nil {
+		// Same predicate as the ready payload and reconnect replay filtering.
+		allowed, err = h.computeAllowedChannels(ctx, h.db, user)
+	}
+	if err != nil || user == nil {
+		// Visibility unresolved. Keeping the old subscriptions would leak, and
+		// revoking them all would hollow out a sidebar the user may still be
+		// entitled to, so close the socket instead: the client reconnects and
+		// rebuilds from a ready payload computed with the new role. kickClient
+		// rather than DisconnectUser — the latter sends a BANNED error, which
+		// makes the client clear its credentials instead of reconnecting.
+		slog.Warn("hub: role change visibility unresolved, closing socket",
+			"user_id", userID, "err", err)
+		h.kickClient(c)
+		return
+	}
+
+	for _, topic := range h.pubsub.TopicsForClient(userID) {
+		chID := channelTopicID(topic)
+		if chID == 0 || allowed[chID] {
+			continue
+		}
+		// DM access is gated on dm_participants, which no role change can
+		// alter, while allowed sources DMs from dm_open_state — a DM the user
+		// has closed (or every DM, if the DM lookup inside
+		// computeAllowedChannels failed) is missing from allowed even though
+		// its subscription is still legitimate. Never revoke a DM topic here;
+		// on a lookup error close the socket rather than guess.
+		ch, chErr := h.db.GetChannel(ctx, chID)
+		if chErr != nil {
+			slog.Warn("hub: role change channel lookup failed, closing socket",
+				"user_id", userID, "channel_id", chID, "err", chErr)
+			h.kickClient(c)
+			return
+		}
+		if ch != nil && ch.Type == "dm" {
+			continue
+		}
+		c.sendMsg(buildChannelDelete(chID))
+		h.pubsub.Unsubscribe(c, topic)
+		c.mu.Lock()
+		if c.channelID == chID {
+			c.channelID = 0
+		}
+		c.mu.Unlock()
+	}
 }
 
 // SendToUser delivers msg directly to the client identified by userID.

Server/ws/hub_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"sync"
 	"testing"
 	"testing/fstest"
@@ -1039,6 +1040,116 @@ func TestRefreshChannelVisibility_ForcesFullResyncForStaleResumes(t *testing.T)
 	}
 }
 
+// ─── BroadcastMemberUpdate (role reassignment) ────────────────────────────────
+
+// A role change must drop the live channel topics the new role cannot READ —
+// the subscription is created once, at channel_focus, and otherwise outlives
+// the authorization it was granted under. DM topics are membership-gated, not
+// role-gated, so they must survive.
+func TestBroadcastMemberUpdate_RevokesUnreadableSubscriptions(t *testing.T) {
+	hub, database := newTestHub(t)
+	go hub.Run()
+	defer hub.Stop()
+	ctx := context.Background()
+
+	chID := seedTestChannel(t, database, "role-room")
+	dmID, err := database.CreateChannel(ctx, "dm-room", "dm", "", "", 0)
+	if err != nil {
+		t.Fatalf("CreateChannel(dm): %v", err)
+	}
+
+	// Connect as Owner (reads everything) with the text channel focused.
+	user := seedOwnerUser(t, database, "demote-me")
+	if _, err := database.ExecContext(ctx,
+		`INSERT INTO dm_participants (channel_id, user_id) VALUES (?, ?)`, dmID, user.ID,
+	); err != nil {
+		t.Fatalf("insert dm_participants: %v", err)
+	}
+	send := make(chan []byte, 16)
+	c := ws.NewTestClientWithUser(hub, user, chID, send)
+	hub.Register(c)
+	time.Sleep(30 * time.Millisecond)
+	// The user is a participant of this DM but has closed it (no dm_open_state
+	// row), so the topic is held while being absent from the allowed set.
+	hub.PubSubForTest().Subscribe(c, ws.ChannelTopic(dmID))
+
+	// Demote to Member, with READ_MESSAGES denied to that role on the channel.
+	if _, err := database.ExecContext(ctx,
+		`INSERT INTO channel_overrides (channel_id, role_id, allow, deny) VALUES (?, 4, 0, 2)`, chID,
+	); err != nil {
+		t.Fatalf("insert override: %v", err)
+	}
+	if err := database.UpdateUserRole(ctx, user.ID, 4); err != nil {
+		t.Fatalf("UpdateUserRole: %v", err)
+	}
+
+	hub.SeedSeq(41)
+	hub.BroadcastMemberUpdate(user.ID, "member")
+
+	msg := drainForMsgType(t, send, "channel_delete")
+	payload, _ := msg["payload"].(map[string]any)
+	id, _ := payload["id"].(float64)
+	if int64(id) != chID {
+		t.Errorf("channel_delete id = %v, want %d", payload["id"], chID)
+	}
+
+	// The socket stays up; only the unreadable topic is gone.
+	if got := hub.ClientCount(); got != 1 {
+		t.Fatalf("ClientCount = %d, want 1 (socket must stay up)", got)
+	}
+	topics := hub.PubSubForTest().TopicsForClient(user.ID)
+	if slices.Contains(topics, ws.ChannelTopic(chID)) {
+		t.Error("still subscribed to the revoked channel topic")
+	}
+	if !slices.Contains(topics, ws.ChannelTopic(dmID)) {
+		t.Error("DM topic revoked by a role change (DM access is membership-gated)")
+	}
+
+	// The impact itself: channel traffic no longer reaches the demoted socket.
+	hub.BroadcastToChannel(chID, []byte(`{"type":"chat_message"}`))
+	time.Sleep(30 * time.Millisecond)
+	assertNoMsgType(t, send, "chat_message")
+
+	// A resume across the change must take the full-ready path.
+	if !hub.MustFullResyncForTest(41) {
+		t.Error("expected forced full resync for a resume at the change watermark")
+	}
+}
+
+// When the new visibility cannot be resolved (DB hiccup), the socket is closed
+// rather than left half-revoked: the client reconnects and rebuilds from a
+// ready payload computed with the new role.
+func TestBroadcastMemberUpdate_ClosesSocketWhenVisibilityUnresolved(t *testing.T) {
+	hub, database := newTestHub(t)
+	go hub.Run()
+	defer hub.Stop()
+	ctx := context.Background()
+
+	chID := seedTestChannel(t, database, "hiccup-room")
+	uid := seedTestUser(t, database, "hiccup-user")
+	user, err := database.GetUserByID(ctx, uid)
+	if err != nil || user == nil {
+		t.Fatalf("GetUserByID: %v", err)
+	}
+	send := make(chan []byte, 16)
+	hub.Register(ws.NewTestClientWithUser(hub, user, chID, send))
+	time.Sleep(30 * time.Millisecond)
+
+	// Break the override lookup computeAllowedChannels depends on.
+	if _, err := database.ExecContext(ctx, `DROP TABLE channel_overrides`); err != nil {
+		t.Fatalf("drop channel_overrides: %v", err)
+	}
+
+	hub.BroadcastMemberUpdate(uid, "member")
+
+	if got := hub.ClientCount(); got != 0 {
+		t.Errorf("ClientCount = %d, want 0 (socket must close when visibility is unresolved)", got)
+	}
+	if topics := hub.PubSubForTest().TopicsForClient(uid); len(topics) != 0 {
+		t.Errorf("TopicsForClient = %v, want none", topics)
+	}
+}
+
 // hubTestSchema is the minimal schema needed for hub tests.
 var hubTestSchema = []byte(`
 CREATE TABLE IF NOT EXISTS roles (

Server/ws/pubsub.go

@@ -2,6 +2,8 @@ package ws
 
 import (
 	"fmt"
+	"strconv"
+	"strings"
 	"sync"
 )
 
@@ -22,6 +24,20 @@ func ChannelTopic(channelID int64) Topic {
 	return Topic(fmt.Sprintf("channel:%d", channelID))
 }
 
+// channelTopicID is the inverse of ChannelTopic: it returns the channel ID
+// encoded in t, or 0 if t is not a text-channel topic.
+func channelTopicID(t Topic) int64 {
+	rest, ok := strings.CutPrefix(string(t), "channel:")
+	if !ok {
+		return 0
+	}
+	id, err := strconv.ParseInt(rest, 10, 64)
+	if err != nil {
+		return 0
+	}
+	return id
+}
+
 // VoiceTopic returns the topic for a voice channel.
 func VoiceTopic(channelID int64) Topic {
 	return Topic(fmt.Sprintf("voice:%d", channelID))