Push a live signal when a user is added to a room (#26)

Previously GET /api/rooms/mine was only ever fetched once at app mount,
so a room added mid-session stayed invisible until a full page reload
-- add_member had no way to reach an already-open client at all.

Backend: ConnectionManager and Broadcaster (renamed from RoomBroadcaster)
now support per-user channels alongside the existing per-room ones, so a
signal can reach a user's socket even for a room they haven't joined
(and by definition can't have, until this fires). add_member publishes
a room_added event on the target user's channel.

Frontend: the WebSocket connection is no longer scoped to whichever
room is open -- ChatShellPage now owns one persistent connection for
the whole session (including while no room is open, which is exactly
when this bug showed), and ChatPane joins/leaves rooms on top of it.
A room_added event triggers a room-list refetch with no reload needed.

Verified end-to-end in the browser: a user sitting on the empty room
list saw a newly-added room appear live, then chatted in it normally.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/main.py

@@ -12,7 +12,7 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from app.config import settings
 from app.routers import admin, auth, bots, health, push, rooms, signup, uploads, users, webhooks
-from app.ws.broadcaster import RoomBroadcaster
+from app.ws.broadcaster import Broadcaster
 from app.ws.chat import router as ws_router
 from app.ws.connection_manager import ConnectionManager
 from app.ws.presence import Presence
@@ -68,7 +68,7 @@ def create_app() -> FastAPI:
     app.state.connection_manager = ConnectionManager()
     app.state.redis = Redis.from_url(settings.redis_url, decode_responses=True)
     app.state.presence = Presence(app.state.redis)
-    app.state.broadcaster = RoomBroadcaster(app.state.redis, app.state.connection_manager)
+    app.state.broadcaster = Broadcaster(app.state.redis, app.state.connection_manager)
 
     app.include_router(health.router)
     app.include_router(auth.router)

backend/app/routers/rooms.py

@@ -34,6 +34,7 @@ from app.schemas.webhook import (
     WebhookIncomingCreate,
     WebhookIncomingRead,
 )
+from app.services.message_events import broadcast_room_added
 from app.services.message_service import get_reactions_for_messages, list_recent_messages
 from app.services.upload_settings_service import format_mb, get_upload_settings
 from app.services.room_service import (
@@ -466,6 +467,7 @@ async def add_member_endpoint(
         raise HTTPException(status_code=404, detail="No user with that ID")
     except AlreadyMemberError:
         raise HTTPException(status_code=409, detail="That user is already a member")
+    await broadcast_room_added(request.app.state.broadcaster, data.user_id, room)
     return RoomMemberRead(
         user_id=membership.user_id,
         username=membership.user.username,

backend/app/services/message_events.py

@@ -7,7 +7,7 @@ from app.models import Message, MessageFile, Room, RoomMembership, User
 from app.schemas.message import ReactionSummary
 from app.services.push_service import send_push_to_user
 from app.services.webhook_service import dispatch_event
-from app.ws.broadcaster import RoomBroadcaster
+from app.ws.broadcaster import Broadcaster
 from app.ws.presence import Presence
 
 
@@ -70,7 +70,7 @@ async def _message_payload(db: AsyncSession, message: Message, username: str) ->
 
 async def broadcast_new_message(
     db: AsyncSession,
-    broadcaster: RoomBroadcaster,
+    broadcaster: Broadcaster,
     presence: Presence,
     room_id: uuid.UUID,
     message: Message,
@@ -86,7 +86,7 @@ async def broadcast_new_message(
 
 
 async def broadcast_message_update(
-    db: AsyncSession, broadcaster: RoomBroadcaster, room_id: uuid.UUID, message: Message
+    db: AsyncSession, broadcaster: Broadcaster, room_id: uuid.UUID, message: Message
 ) -> None:
     payload = {
         "type": "message_update",
@@ -100,7 +100,7 @@ async def broadcast_message_update(
 
 
 async def broadcast_reaction_update(
-    broadcaster: RoomBroadcaster,
+    broadcaster: Broadcaster,
     room_id: uuid.UUID,
     message_id: uuid.UUID,
     reactions: list[ReactionSummary],
@@ -115,3 +115,15 @@ async def broadcast_reaction_update(
     # Deliberately no dispatch_event() call -- reactions don't get an
     # outgoing-webhook event type, matching the same scope cut made for
     # image uploads (see backend/README.md).
+
+
+async def broadcast_room_added(broadcaster: Broadcaster, user_id: uuid.UUID, room: Room) -> None:
+    """The only signal a user's open client gets that they were just added
+    to a room -- without it, GET /rooms/mine is only ever fetched once at
+    app mount, so a room added mid-session stays invisible until a full
+    reload. Published on the user's own channel rather than the room's,
+    since the whole point is reaching someone who hasn't joined that room's
+    channel yet (and by definition can't have)."""
+    await broadcaster.publish_to_user(
+        user_id, {"type": "room_added", "room_id": str(room.id)}
+    )

backend/app/ws/broadcaster.py

@@ -6,16 +6,24 @@ from redis.asyncio import Redis
 from app.ws.connection_manager import ConnectionManager
 
 ROOM_CHANNEL_PREFIX = "room:"
+USER_CHANNEL_PREFIX = "user:"
 
 
-class RoomBroadcaster:
+class Broadcaster:
     """Cross-instance message fan-out (ARCHITECTURE.md phase 5).
 
-    Publishes to a per-room Redis channel; every app instance -- including
+    Publishes to a per-room or per-user Redis channel; every app instance --
-    the one that published -- subscribes via a single pattern subscription
+    including the one that published -- subscribes via a single pattern
-    and forwards to its own locally connected WebSocket clients via
+    subscription and forwards to its own locally connected WebSocket clients
-    ConnectionManager. A single instance just talks to itself through Redis,
+    via ConnectionManager. A single instance just talks to itself through
-    so there's no separate code path for the 1-instance vs N-instance case.
+    Redis, so there's no separate code path for the 1-instance vs N-instance
+    case.
+
+    Room channels carry anything scoped to a room's joined members (new
+    messages, edits, reactions). User channels carry anything scoped to one
+    person regardless of which rooms they've joined -- currently just
+    "you've been added to a room," which by definition arrives before the
+    recipient could ever have joined that room's own channel.
     """
 
     def __init__(self, redis: Redis, manager: ConnectionManager) -> None:
@@ -25,17 +33,24 @@ class RoomBroadcaster:
     async def publish(self, room_id: uuid.UUID, payload: dict) -> None:
         await self._redis.publish(f"{ROOM_CHANNEL_PREFIX}{room_id}", json.dumps(payload))
 
+    async def publish_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
+        await self._redis.publish(f"{USER_CHANNEL_PREFIX}{user_id}", json.dumps(payload))
+
     async def listen(self) -> None:
         pubsub = self._redis.pubsub()
-        await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*")
+        await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
         try:
             async for message in pubsub.listen():
                 if message["type"] != "pmessage":
                     continue
                 channel = message["channel"]
-                room_id = uuid.UUID(channel.removeprefix(ROOM_CHANNEL_PREFIX))
                 payload = json.loads(message["data"])
-                await self._manager.broadcast(room_id, payload)
+                if channel.startswith(ROOM_CHANNEL_PREFIX):
+                    room_id = uuid.UUID(channel.removeprefix(ROOM_CHANNEL_PREFIX))
+                    await self._manager.broadcast(room_id, payload)
+                elif channel.startswith(USER_CHANNEL_PREFIX):
+                    user_id = uuid.UUID(channel.removeprefix(USER_CHANNEL_PREFIX))
+                    await self._manager.send_to_user(user_id, payload)
         finally:
-            await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*")
+            await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
             await pubsub.aclose()

backend/app/ws/chat.py

@@ -78,6 +78,7 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
     presence = websocket.app.state.presence
     broadcaster = websocket.app.state.broadcaster
     joined_rooms: set[uuid.UUID] = set()
+    manager.register_user(user.id, websocket)
 
     try:
         while True:
@@ -225,5 +226,6 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
         pass
     finally:
         manager.leave_all(websocket)
+        manager.unregister_user(user.id, websocket)
         for room_id in joined_rooms:
             await presence.leave(room_id, user.id)

backend/app/ws/connection_manager.py

@@ -8,13 +8,14 @@ class ConnectionManager:
     """Local, single-process WebSocket socket registry.
 
     Purely about delivering to sockets connected to *this* process --
-    cross-instance fan-out lives in RoomBroadcaster, and cross-instance
+    cross-instance fan-out lives in Broadcaster, and cross-instance "who's
-    "who's connected" for push lives in Presence, both backed by Redis
+    connected" for push lives in Presence, both backed by Redis
     (ARCHITECTURE.md phase 5).
     """
 
     def __init__(self) -> None:
         self._rooms: dict[uuid.UUID, set[WebSocket]] = defaultdict(set)
+        self._users: dict[uuid.UUID, set[WebSocket]] = defaultdict(set)
 
     def join(self, room_id: uuid.UUID, websocket: WebSocket) -> None:
         self._rooms[room_id].add(websocket)
@@ -28,6 +29,22 @@ class ConnectionManager:
         for room_id in list(self._rooms.keys()):
             self.leave(room_id, websocket)
 
+    def register_user(self, user_id: uuid.UUID, websocket: WebSocket) -> None:
+        """Ties a socket to the authenticated user who owns it, independent
+        of which (if any) room it has joined -- lets a user be reached the
+        instant they're added to a room, before they've ever joined that
+        room's channel."""
+        self._users[user_id].add(websocket)
+
+    def unregister_user(self, user_id: uuid.UUID, websocket: WebSocket) -> None:
+        self._users[user_id].discard(websocket)
+        if not self._users[user_id]:
+            del self._users[user_id]
+
     async def broadcast(self, room_id: uuid.UUID, payload: dict) -> None:
         for websocket in list(self._rooms.get(room_id, ())):
             await websocket.send_json(payload)
+
+    async def send_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
+        for websocket in list(self._users.get(user_id, ())):
+            await websocket.send_json(payload)

backend/tests/test_broadcast.py

@@ -8,6 +8,16 @@ def _unique(prefix: str) -> str:
     return f"{prefix}-{uuid.uuid4().hex[:8]}"
 
 
+def _fake_send_email(monkeypatch):
+    calls = []
+
+    async def fake(db, to, subject, body):
+        calls.append({"to": to, "subject": subject, "body": body})
+
+    monkeypatch.setattr("app.services.room_service.send_email", fake)
+    return calls
+
+
 def _register_ws(ws_client, username: str) -> dict:
     async def _seed():
         async with ws_client.session_factory() as session:
@@ -101,3 +111,26 @@ def test_presence_is_shared_across_instances(ws_client_factory, monkeypatch):
             assert alice_ws.receive_json()["type"] == "joined"
 
     assert calls == []
+
+
+def test_add_member_notifies_target_user_via_websocket(ws_client_factory, monkeypatch):
+    # Bob is only ever "connected," never "joined" -- proving the room_added
+    # signal reaches him on his own per-user channel, independent of (and
+    # necessarily before) ever joining the room's own channel, which he
+    # can't do until this signal tells his client the room exists at all.
+    _fake_send_email(monkeypatch)
+
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    alice = _register_ws(instance1, _unique("alice"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+
+    bob = _register_ws(instance2, _unique("bob"))
+
+    with instance2.websocket_connect("/ws/chat") as bob_ws:
+        resp = instance1.post(f"/api/rooms/{room['id']}/members", json={"user_id": bob["id"]})
+        assert resp.status_code == 201, resp.text
+
+        received = bob_ws.receive_json()
+        assert received == {"type": "room_added", "room_id": room["id"]}

frontend/src/components/ChatPane.tsx

@@ -1,8 +1,7 @@
 import { useCallback, useEffect, useState } from 'react'
-import { useNavigate } from 'react-router-dom'
 import { NetworkError } from '../api/client'
 import { getRoomMessages } from '../api/rooms'
-import { useChatSocket } from '../ws/useChatSocket'
+import type { ChatSocketHandle } from '../ws/useChatSocket'
 import type { ChatMessageEnvelope, Message, MyRoomItem, RoomMember, ServerEnvelope } from '../types'
 import { Composer } from './Composer'
 import { MessageList } from './MessageList'
@@ -15,10 +14,10 @@ interface ChatPaneProps {
   onBack: () => void
   onToggleInfo: () => void
   infoOpen: boolean
+  socket: ChatSocketHandle
 }
 
-export function ChatPane({ room, members, isMobile, onBack, onToggleInfo, infoOpen }: ChatPaneProps) {
+export function ChatPane({ room, members, isMobile, onBack, onToggleInfo, infoOpen, socket }: ChatPaneProps) {
-  const navigate = useNavigate()
   const [history, setHistory] = useState<Message[]>([])
   const [live, setLive] = useState<ChatMessageEnvelope[]>([])
   const [wsError, setWsError] = useState<string | null>(null)
@@ -40,35 +39,60 @@ export function ChatPane({ room, members, isMobile, onBack, onToggleInfo, infoOp
       })
   }, [room.id])
 
-  const onMessage = useCallback((envelope: ServerEnvelope) => {
+  useEffect(() => {
-    if (envelope.type === 'message') {
+    socket.joinRoom(room.id)
-      setLive((prev) => [...prev, envelope])
+    return () => socket.leaveRoom(room.id)
-    } else if (envelope.type === 'message_update') {
+  }, [socket, room.id])
-      setHistory((prev) =>
-        prev.map((m) =>
-          m.id === envelope.id ? { ...m, content: envelope.content, edited_at: envelope.edited_at } : m,
-        ),
-      )
-      setLive((prev) =>
-        prev.map((m) =>
-          m.id === envelope.id ? { ...m, content: envelope.content, edited_at: envelope.edited_at } : m,
-        ),
-      )
-    } else if (envelope.type === 'reaction_update') {
-      setHistory((prev) =>
-        prev.map((m) => (m.id === envelope.id ? { ...m, reactions: envelope.reactions } : m)),
-      )
-      setLive((prev) =>
-        prev.map((m) => (m.id === envelope.id ? { ...m, reactions: envelope.reactions } : m)),
-      )
-    } else if (envelope.type === 'error') {
-      setWsError(envelope.detail)
-    }
-  }, [])
 
-  const onUnauthenticated = useCallback(() => navigate('/login'), [navigate])
+  useEffect(
+    () =>
+      // The socket is shared across every room this tab visits, so a
+      // stray in-flight event for a room just left (or a different tab's
+      // room, in theory) has to be filtered out here rather than assumed
+      // away -- `error` has no room_id to filter on, but is rare enough
+      // that misattributing one to the wrong room's banner isn't worth
+      // guarding against separately.
+      socket.subscribe((envelope: ServerEnvelope) => {
+        if (envelope.type === 'message' && envelope.room_id === room.id) {
+          setLive((prev) => [...prev, envelope])
+        } else if (envelope.type === 'message_update' && envelope.room_id === room.id) {
+          setHistory((prev) =>
+            prev.map((m) =>
+              m.id === envelope.id ? { ...m, content: envelope.content, edited_at: envelope.edited_at } : m,
+            ),
+          )
+          setLive((prev) =>
+            prev.map((m) =>
+              m.id === envelope.id ? { ...m, content: envelope.content, edited_at: envelope.edited_at } : m,
+            ),
+          )
+        } else if (envelope.type === 'reaction_update' && envelope.room_id === room.id) {
+          setHistory((prev) =>
+            prev.map((m) => (m.id === envelope.id ? { ...m, reactions: envelope.reactions } : m)),
+          )
+          setLive((prev) =>
+            prev.map((m) => (m.id === envelope.id ? { ...m, reactions: envelope.reactions } : m)),
+          )
+        } else if (envelope.type === 'error') {
+          setWsError(envelope.detail)
+        }
+      }),
+    [socket, room.id],
+  )
 
-  const { connected, send, sendEdit, sendReaction } = useChatSocket({ roomId: room.id, onMessage, onUnauthenticated })
+  const connected = socket.connected
+  const send = useCallback(
+    (content: string, imageId?: string, fileId?: string) => socket.send(room.id, content, imageId, fileId),
+    [socket, room.id],
+  )
+  const sendEdit = useCallback(
+    (messageId: string, content: string) => socket.sendEdit(room.id, messageId, content),
+    [socket, room.id],
+  )
+  const sendReaction = useCallback(
+    (messageId: string, emoji: string) => socket.sendReaction(room.id, messageId, emoji),
+    [socket, room.id],
+  )
 
   return (
     <section className="chat-pane">

frontend/src/pages/ChatShellPage.tsx

@@ -12,6 +12,7 @@ import { TopBar } from '../components/TopBar'
 import { useAuth } from '../context/AuthContext'
 import { MOBILE_BREAKPOINT, useWindowWidth } from '../hooks/useWindowWidth'
 import type { MyRoomItem, RoomMember } from '../types'
+import { useChatSocket } from '../ws/useChatSocket'
 import './ChatShellPage.css'
 
 type ModalKind = 'new' | 'browse' | null
@@ -56,6 +57,17 @@ export function ChatShellPage() {
     refreshRooms().catch(() => {})
   }, [refreshRooms])
 
+  const onSocketUnauthenticated = useCallback(() => navigate('/login'), [navigate])
+  const socket = useChatSocket({ onUnauthenticated: onSocketUnauthenticated })
+
+  useEffect(
+    () =>
+      socket.subscribe((envelope) => {
+        if (envelope.type === 'room_added') refreshRooms()
+      }),
+    [socket, refreshRooms],
+  )
+
   useEffect(() => {
     refreshMembers()
     // Also re-run when the logged-in user's own profile changes (display
@@ -97,6 +109,7 @@ export function ChatShellPage() {
               onBack={() => navigate('/rooms')}
               onToggleInfo={() => setInfoOpen((v) => !v)}
               infoOpen={infoOpen}
+              socket={socket}
             />
           ) : (
             !isMobile && (

frontend/src/types.ts

@@ -114,12 +114,18 @@ export interface ChatErrorEnvelope {
   detail: string
 }
 
+export interface ChatRoomAddedEnvelope {
+  type: 'room_added'
+  room_id: string
+}
+
 export type ServerEnvelope =
   | ChatMessageEnvelope
   | ChatMessageUpdateEnvelope
   | ChatReactionUpdateEnvelope
   | ChatJoinedEnvelope
   | ChatErrorEnvelope
+  | ChatRoomAddedEnvelope
 
 export interface AdminUser {
   id: string

frontend/src/ws/useChatSocket.ts

@@ -2,21 +2,25 @@ import { useCallback, useEffect, useRef, useState } from 'react'
 import type { ServerEnvelope } from '../types'
 
 interface UseChatSocketOptions {
-  roomId: string
-  onMessage: (envelope: ServerEnvelope) => void
   onUnauthenticated: () => void
 }
 
 const RECONNECT_BASE_DELAY_MS = 1000
 const RECONNECT_MAX_DELAY_MS = 30000
 
-export function useChatSocket({ roomId, onMessage, onUnauthenticated }: UseChatSocketOptions) {
+// One connection per authenticated session, established as soon as the app
+// shell mounts -- not per-room. A room is just something this socket can be
+// told to "join"/"leave" while it's open; the connection itself persists
+// across room switches and while no room is open at all, since a per-user
+// signal (e.g. "you were added to a room") has to reach the client whether
+// or not any room is currently open.
+export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
   const socketRef = useRef<WebSocket | null>(null)
   const [connected, setConnected] = useState(false)
-  const onMessageRef = useRef(onMessage)
-  onMessageRef.current = onMessage
   const onUnauthenticatedRef = useRef(onUnauthenticated)
   onUnauthenticatedRef.current = onUnauthenticated
+  const subscribersRef = useRef(new Set<(envelope: ServerEnvelope) => void>())
+  const joinedRoomsRef = useRef(new Set<string>())
 
   useEffect(() => {
     let stopped = false
@@ -38,12 +42,17 @@ export function useChatSocket({ roomId, onMessage, onUnauthenticated }: UseChatS
         if (socketRef.current !== ws) return
         reconnectDelay = RECONNECT_BASE_DELAY_MS
         setConnected(true)
-        ws.send(JSON.stringify({ type: 'join', room_id: roomId }))
+        // Re-join whatever rooms were joined before a reconnect -- the
+        // server has no memory of a dropped connection's prior state.
+        for (const roomId of joinedRoomsRef.current) {
+          ws.send(JSON.stringify({ type: 'join', room_id: roomId }))
+        }
       }
 
       ws.onmessage = (event) => {
         if (socketRef.current !== ws) return
-        onMessageRef.current(JSON.parse(event.data) as ServerEnvelope)
+        const envelope = JSON.parse(event.data) as ServerEnvelope
+        for (const handler of subscribersRef.current) handler(envelope)
       }
 
       ws.onclose = (event) => {
@@ -80,9 +89,32 @@ export function useChatSocket({ roomId, onMessage, onUnauthenticated }: UseChatS
       socketRef.current?.close()
       socketRef.current = null
     }
-  }, [roomId])
+  }, [])
 
-  const send = useCallback((content: string, imageId?: string, fileId?: string) => {
+  const subscribe = useCallback((handler: (envelope: ServerEnvelope) => void) => {
+    subscribersRef.current.add(handler)
+    return () => {
+      subscribersRef.current.delete(handler)
+    }
+  }, [])
+
+  const joinRoom = useCallback((roomId: string) => {
+    joinedRoomsRef.current.add(roomId)
+    const ws = socketRef.current
+    if (ws && ws.readyState === WebSocket.OPEN) {
+      ws.send(JSON.stringify({ type: 'join', room_id: roomId }))
+    }
+  }, [])
+
+  const leaveRoom = useCallback((roomId: string) => {
+    joinedRoomsRef.current.delete(roomId)
+    const ws = socketRef.current
+    if (ws && ws.readyState === WebSocket.OPEN) {
+      ws.send(JSON.stringify({ type: 'leave', room_id: roomId }))
+    }
+  }, [])
+
+  const send = useCallback((roomId: string, content: string, imageId?: string, fileId?: string) => {
     const ws = socketRef.current
     if (!ws || ws.readyState !== WebSocket.OPEN) return
     ws.send(
@@ -94,19 +126,21 @@ export function useChatSocket({ roomId, onMessage, onUnauthenticated }: UseChatS
         file_id: fileId ?? null,
       }),
     )
-  }, [roomId])
+  }, [])
 
-  const sendEdit = useCallback((messageId: string, content: string) => {
+  const sendEdit = useCallback((roomId: string, messageId: string, content: string) => {
     const ws = socketRef.current
     if (!ws || ws.readyState !== WebSocket.OPEN) return
     ws.send(JSON.stringify({ type: 'edit', room_id: roomId, message_id: messageId, content }))
-  }, [roomId])
+  }, [])
 
-  const sendReaction = useCallback((messageId: string, emoji: string) => {
+  const sendReaction = useCallback((roomId: string, messageId: string, emoji: string) => {
     const ws = socketRef.current
     if (!ws || ws.readyState !== WebSocket.OPEN) return
     ws.send(JSON.stringify({ type: 'reaction', room_id: roomId, message_id: messageId, emoji }))
-  }, [roomId])
+  }, [])
 
-  return { connected, send, sendEdit, sendReaction }
+  return { connected, subscribe, joinRoom, leaveRoom, send, sendEdit, sendReaction }
 }
+
+export type ChatSocketHandle = ReturnType<typeof useChatSocket>