maxim.dolgolyov/Learn_System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move emitToSession broadcast into ws-server.broadcastToSession

Browse Source 
Consolidates WS-first + SSE-fallback + guest forwarding logic
into ws-server.js so classroomController doesn't duplicate delivery.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Maxim Dolgolyov
2026-04-13 21:27:14 +03:00
parent edb4c211a0
commit 0ac292ab9e
2 changed files with 232 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/controllers/classroomController.js
+3 -15
View File
@@ -25,22 +25,10 @@ const GUEST_EVENTS = new Set([
]);
/* ── Helper: broadcast to all session participants ─────────────────────── */
/* Delivery: WS-first (ws-server.js) with SSE fallback for non-WS users. */
function emitToSession(sessionId, data) {
const session = db.prepare('SELECT class_id, teacher_id FROM classroom_sessions WHERE id=?').get(sessionId);
if (!session) return;
if (session.class_id) {
emitToClass(session.class_id, data);
emit(session.teacher_id, data); // teacher is not in class_members — emit separately
} else {
// personal session — emit to teacher + each invited user
emit(session.teacher_id, data);
const invites = db.prepare('SELECT user_id FROM classroom_invites WHERE session_id=?').all(sessionId);
for (const { user_id } of invites) emit(user_id, data);
}
// Forward whitelisted events to guest viewers
if (GUEST_EVENTS.has(data.type)) emitToGuests(sessionId, data);
// ws-server handles WS-first + SSE-fallback + guest forwarding internally
require('../ws-server').broadcastToSession(sessionId, data, GUEST_EVENTS.has(data.type));
}
/* ── Helper: check if user has access to session ──────────────────────── */
backend/src/ws-server.js
+229 -71
View File
@@ -1,23 +1,68 @@
/**
* WebSocket server for low-latency real-time classroom events.
* WebSocket server — full real-time classroom channel.
*
* Handles two message types from clients:
* { type:'cursor', sessionId, x, y, pageNum }
* { type:'preview', sessionId, liveId, tool, data, pageNum, cancel }
* Client → server messages:
* classroom_join { sessionId }
* cursor { sessionId, x, y, pageNum }
* preview { sessionId, liveId, tool, data, pageNum, cancel }
* hand_raise { sessionId }
* hand_lower { sessionId, targetUserId? } teacher can lower anyone's hand
* page_change { sessionId, pageNum } teacher only
* page_clear { sessionId, pageNum } teacher only
* template_change { sessionId, pageNum, template } teacher only
* allow_draw { sessionId, targetUserId } teacher only
* revoke_draw { sessionId, targetUserId } teacher only
* mute_peer { sessionId, targetUserId } teacher only
* screen_start { sessionId } teacher only
* screen_stop { sessionId } teacher only
*
* Auth: JWT token in ?token= query param on upgrade request.
* Forwards events via SSE to session participants (no DB write).
*
* This replaces the HTTP POST /cursor and /stroke-preview endpoints
* for connected clients, reducing per-event latency from ~50-120ms to ~2-5ms.
* Server → client delivery:
* All classroom events are sent via WS to connected users,
* with automatic SSE fallback for users not on WS.
* Use broadcastToSession() / emitToUser() from other modules.
*/
const { WebSocketServer } = require('ws');
const jwt = require('jsonwebtoken');
const db = require('./db/db');
const { emit, emitToGuests } = require('./sse');
/* ── Session member cache (avoids DB query per WS message) ────────────── */
const _cache = new Map(); // sessionId → { teacherId, classId, userIds, ts }
/* ── Classroom connections: userId → Set<ws> ──────────────────────────── */
const _classroomConns = new Map();
function _registerUser(ws) {
if (!_classroomConns.has(ws.userId)) _classroomConns.set(ws.userId, new Set());
_classroomConns.get(ws.userId).add(ws);
}
function _unregisterUser(ws) {
const set = _classroomConns.get(ws.userId);
if (!set) return;
set.delete(ws);
if (set.size === 0) _classroomConns.delete(ws.userId);
}
/* ── Delivery: WS first, SSE fallback ────────────────────────────────── */
function emitToUser(userId, data) {
const sockets = _classroomConns.get(userId);
if (sockets?.size) {
const raw = JSON.stringify(data);
for (const ws of sockets) {
if (ws.readyState === 1) try { ws.send(raw); } catch {}
}
return; // delivered via WS
}
emit(userId, data); // SSE fallback
}
function broadcastToSession(sessionId, data, includeGuests = false) {
const members = _getMembers(sessionId);
if (!members) return;
for (const uid of members.userIds) emitToUser(uid, data);
if (includeGuests) emitToGuests(sessionId, data);
}
/* ── Session member cache (30s TTL — avoids DB per message) ───────────── */
const _cache = new Map();
const CACHE_TTL = 30_000;
function _getMembers(sessionId) {
@@ -31,11 +76,11 @@ function _getMembers(sessionId) {
let userIds;
if (session.class_id) {
const members = db.prepare('SELECT user_id FROM class_members WHERE class_id=?').all(session.class_id);
userIds = [session.teacher_id, ...members.map(m => m.user_id)];
const rows = db.prepare('SELECT user_id FROM class_members WHERE class_id=?').all(session.class_id);
userIds = [session.teacher_id, ...rows.map(r => r.user_id)];
} else {
const invites = db.prepare('SELECT user_id FROM classroom_invites WHERE session_id=?').all(sessionId);
userIds = [session.teacher_id, ...invites.map(i => i.user_id)];
const rows = db.prepare('SELECT user_id FROM classroom_invites WHERE session_id=?').all(sessionId);
userIds = [session.teacher_id, ...rows.map(r => r.user_id)];
}
const entry = { teacherId: session.teacher_id, classId: session.class_id, userIds, ts: Date.now() };
@@ -43,21 +88,11 @@ function _getMembers(sessionId) {
return entry;
}
function _invalidateSession(sessionId) {
_cache.delete(sessionId);
}
function _invalidateSession(sessionId) { _cache.delete(sessionId); }
/* Forward serialized SSE payload to all session members */
function _broadcast(sessionId, data, includeGuests) {
const members = _getMembers(sessionId);
if (!members) return;
for (const uid of members.userIds) emit(uid, data);
if (includeGuests) emitToGuests(sessionId, data);
}
/* Check draw permissions (teacher always can; students need explicit grant) */
const _drawCache = new Map(); // `${sessionId}:${userId}` → { allowed, ts }
const DRAW_TTL = 10_000;
/* ── Draw permission cache (10s TTL) ─────────────────────────────────── */
const _drawCache = new Map();
const DRAW_TTL = 10_000;
function _canDraw(sessionId, userId, members) {
if (!members) return false;
@@ -72,6 +107,166 @@ function _canDraw(sessionId, userId, members) {
return allowed;
}
function _invalidateDrawCache(sessionId, userId) {
_drawCache.delete(`${sessionId}:${userId}`);
}
/* ── Message handler ─────────────────────────────────────────────────── */
function _handleMessage(ws, msg) {
const { type, sessionId } = msg;
if (!sessionId || typeof sessionId !== 'number') return;
const members = _getMembers(sessionId);
if (!members) return;
const isMember = members.userIds.includes(ws.userId);
const isTeacher = members.teacherId === ws.userId;
switch (type) {
/* ── Register in session (sends WS events instead of SSE) ── */
case 'classroom_join':
ws.classroomSessionId = sessionId;
_registerUser(ws);
break;
/* ── Cursor position ── */
case 'cursor': {
if (!isMember) return;
broadcastToSession(sessionId, {
type: 'classroom_cursor', sessionId,
x: msg.x, y: msg.y, pageNum: msg.pageNum || 1,
userId: ws.userId, userName: ws.userName,
}, true);
break;
}
/* ── Stroke live preview ── */
case 'preview': {
if (!_canDraw(sessionId, ws.userId, members)) return;
if (!msg.liveId && !msg.cancel) return;
broadcastToSession(sessionId, {
type: 'classroom_stroke_preview', sessionId,
pageNum: msg.pageNum || 1, liveId: msg.liveId,
tool: msg.tool, data: msg.data, cancel: msg.cancel || false,
userId: ws.userId, userName: ws.userName,
}, true);
break;
}
/* ── Hand raise / lower ── */
case 'hand_raise': {
if (!isMember) return;
try {
db.prepare('INSERT OR IGNORE INTO classroom_hands (session_id, user_id) VALUES (?,?)').run(sessionId, ws.userId);
} catch { return; }
broadcastToSession(sessionId, {
type: 'classroom_hand_raised', sessionId,
userId: ws.userId, userName: ws.userName,
});
break;
}
case 'hand_lower': {
if (!isMember) return;
const targetId = (isTeacher && msg.targetUserId) ? Number(msg.targetUserId) : ws.userId;
try {
db.prepare('DELETE FROM classroom_hands WHERE session_id=? AND user_id=?').run(sessionId, targetId);
} catch { return; }
broadcastToSession(sessionId, {
type: 'classroom_hand_lowered', sessionId, userId: targetId,
});
break;
}
/* ── Page navigation (teacher) ── */
case 'page_change': {
if (!isTeacher) return;
const pageNum = Number(msg.pageNum);
if (!pageNum || pageNum < 1) return;
db.prepare('UPDATE classroom_sessions SET current_page=? WHERE id=?').run(pageNum, sessionId);
broadcastToSession(sessionId, {
type: 'classroom_page_changed', sessionId, pageNum,
}, true);
break;
}
/* ── Page clear (teacher) ── */
case 'page_clear': {
if (!isTeacher) return;
const pageNum = Number(msg.pageNum) || 1;
db.prepare('DELETE FROM classroom_strokes WHERE session_id=? AND page_num=?').run(sessionId, pageNum);
broadcastToSession(sessionId, {
type: 'classroom_page_cleared', sessionId, pageNum,
}, true);
break;
}
/* ── Page template change (teacher) ── */
case 'template_change': {
if (!isTeacher) return;
const pageNum = Number(msg.pageNum) || 1;
const template = (msg.template || 'blank').slice(0, 32);
try {
db.prepare(
`INSERT INTO classroom_pages (session_id, page_num, template) VALUES (?,?,?)
ON CONFLICT(session_id, page_num) DO UPDATE SET template=excluded.template`
).run(sessionId, pageNum, template);
} catch { return; }
broadcastToSession(sessionId, {
type: 'classroom_template_changed', sessionId, pageNum, template,
}, true);
break;
}
/* ── Draw permission (teacher) ── */
case 'allow_draw': {
if (!isTeacher) return;
const targetId = Number(msg.targetUserId);
if (!targetId) return;
try {
db.prepare('INSERT OR IGNORE INTO classroom_draw_permissions (session_id, user_id) VALUES (?,?)').run(sessionId, targetId);
} catch { return; }
_invalidateDrawCache(sessionId, targetId);
emitToUser(targetId, { type: 'classroom_draw_permitted', sessionId });
break;
}
case 'revoke_draw': {
if (!isTeacher) return;
const targetId = Number(msg.targetUserId);
if (!targetId) return;
try {
db.prepare('DELETE FROM classroom_draw_permissions WHERE session_id=? AND user_id=?').run(sessionId, targetId);
} catch { return; }
_invalidateDrawCache(sessionId, targetId);
emitToUser(targetId, { type: 'classroom_draw_revoked', sessionId });
break;
}
/* ── Mute peer (teacher) ── */
case 'mute_peer': {
if (!isTeacher) return;
const targetId = Number(msg.targetUserId);
if (!targetId) return;
emitToUser(targetId, { type: 'classroom_muted', sessionId });
break;
}
/* ── Screen share announce (teacher) ── */
case 'screen_start': {
if (!isTeacher) return;
broadcastToSession(sessionId, { type: 'classroom_screen_started', sessionId });
break;
}
case 'screen_stop': {
if (!isTeacher) return;
broadcastToSession(sessionId, { type: 'classroom_screen_stopped', sessionId });
break;
}
}
}
/* ── WebSocket server ──────────────────────────────────────────────────── */
function attach(httpServer) {
const wss = new WebSocketServer({ server: httpServer, path: '/ws' });
@@ -97,53 +292,16 @@ function attach(httpServer) {
ws.on('message', raw => {
let msg;
try { msg = JSON.parse(raw); } catch { return; }
const { type, sessionId } = msg;
if (!sessionId || typeof sessionId !== 'number') return;
/* ── cursor broadcast ── */
if (type === 'cursor') {
const members = _getMembers(sessionId);
if (!members || !members.userIds.includes(ws.userId)) return;
_broadcast(sessionId, {
type: 'classroom_cursor',
sessionId,
x: msg.x,
y: msg.y,
pageNum: msg.pageNum || 1,
userId: ws.userId,
userName: ws.userName,
}, true);
/* ── stroke preview broadcast ── */
} else if (type === 'preview') {
const members = _getMembers(sessionId);
if (!members) return;
if (!_canDraw(sessionId, ws.userId, members)) return;
const liveId = msg.liveId;
if (!liveId && !msg.cancel) return;
_broadcast(sessionId, {
type: 'classroom_stroke_preview',
sessionId,
pageNum: msg.pageNum || 1,
liveId,
tool: msg.tool,
data: msg.data,
cancel: msg.cancel || false,
userId: ws.userId,
userName: ws.userName,
}, true);
try { _handleMessage(ws, msg); } catch (e) {
// swallow — never crash the WS server on bad input
}
});
ws.on('error', () => {});
ws.on('close', () => {});
ws.on('close', () => { _unregisterUser(ws); });
});
/* ── Ping/pong keepalive ── */
/* ── Ping/pong keepalive (30s) ── */
const pingTimer = setInterval(() => {
for (const ws of wss.clients) {
if (!ws.isAlive) { ws.terminate(); continue; }
@@ -156,4 +314,4 @@ function attach(httpServer) {
return wss;
}
module.exports = { attach, invalidateSession: _invalidateSession };
module.exports = { attach, broadcastToSession, emitToUser, invalidateSession: _invalidateSession };