diff --git a/backend/app/main.py b/backend/app/main.py index 31522a9..a578bff 100644 --- a/backend/app/main.py +++ b/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) diff --git a/backend/app/routers/rooms.py b/backend/app/routers/rooms.py index 6c66e3c..bedcd91 100644 --- a/backend/app/routers/rooms.py +++ b/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, diff --git a/backend/app/services/message_events.py b/backend/app/services/message_events.py index 74f2806..fabdc34 100644 --- a/backend/app/services/message_events.py +++ b/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)} + ) diff --git a/backend/app/ws/broadcaster.py b/backend/app/ws/broadcaster.py index 0beb39d..32f9543 100644 --- a/backend/app/ws/broadcaster.py +++ b/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 - the one that published -- subscribes via a single pattern subscription - and forwards to its own locally connected WebSocket clients via - ConnectionManager. A single instance just talks to itself through Redis, - so there's no separate code path for the 1-instance vs N-instance case. + Publishes to a per-room or per-user Redis channel; every app instance -- + including the one that published -- subscribes via a single pattern + subscription and forwards to its own locally connected WebSocket clients + via ConnectionManager. A single instance just talks to itself through + 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() diff --git a/backend/app/ws/chat.py b/backend/app/ws/chat.py index a68989f..bef28e0 100644 --- a/backend/app/ws/chat.py +++ b/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) diff --git a/backend/app/ws/connection_manager.py b/backend/app/ws/connection_manager.py index 58b770e..08fad8c 100644 --- a/backend/app/ws/connection_manager.py +++ b/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 - "who's connected" for push lives in Presence, both backed by Redis + cross-instance fan-out lives in Broadcaster, and cross-instance "who's + 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) diff --git a/backend/tests/test_broadcast.py b/backend/tests/test_broadcast.py index a9b2d02..6f37db2 100644 --- a/backend/tests/test_broadcast.py +++ b/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"]} diff --git a/frontend/src/components/ChatPane.tsx b/frontend/src/components/ChatPane.tsx index 7c1d5a2..8260128 100644 --- a/frontend/src/components/ChatPane.tsx +++ b/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) { - const navigate = useNavigate() +export function ChatPane({ room, members, isMobile, onBack, onToggleInfo, infoOpen, socket }: ChatPaneProps) { const [history, setHistory] = useState([]) const [live, setLive] = useState([]) const [wsError, setWsError] = useState(null) @@ -40,35 +39,60 @@ export function ChatPane({ room, members, isMobile, onBack, onToggleInfo, infoOp }) }, [room.id]) - const onMessage = useCallback((envelope: ServerEnvelope) => { - if (envelope.type === 'message') { - setLive((prev) => [...prev, envelope]) - } else if (envelope.type === 'message_update') { - 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) - } - }, []) + useEffect(() => { + socket.joinRoom(room.id) + return () => socket.leaveRoom(room.id) + }, [socket, room.id]) - 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 (
diff --git a/frontend/src/pages/ChatShellPage.tsx b/frontend/src/pages/ChatShellPage.tsx index 81680ec..19660eb 100644 --- a/frontend/src/pages/ChatShellPage.tsx +++ b/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 && ( diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 434f9dc..9ba2e8d 100644 --- a/frontend/src/types.ts +++ b/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 diff --git a/frontend/src/ws/useChatSocket.ts b/frontend/src/ws/useChatSocket.ts index 2817662..a7bba92 100644 --- a/frontend/src/ws/useChatSocket.ts +++ b/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(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()) 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