(fixed) latency from 466ms to around 9ms for thousands of concurrent …

apps/ws-server/src/clients/index.ts

@@ -4,3 +4,38 @@ import type { Client } from "../index.js";
 
 export const clients = new Set<Client>();
 export const sessionTimers = new Map<string, NodeJS.Timeout>();
+
+// Clients grouped by session, so a broadcast touches only the people in that
+// quiz instead of scanning every connection. With several quizzes running at
+// once the flat Set means each tick walks all of them; this keeps the work
+// proportional to one session.
+const bySession = new Map<string, Set<Client>>();
+
+/** Clients in one session. Empty set if the session has none. */
+export function clientsInSession(sessionId: string): ReadonlySet<Client> {
+    return bySession.get(sessionId) ?? EMPTY;
+}
+const EMPTY: ReadonlySet<Client> = new Set();
+
+/** Register a client under a session, moving it off any previous one. */
+export function indexClient(client: Client, sessionId: string) {
+    if (client.sessionId && client.sessionId !== sessionId) {
+        unindexClient(client);
+    }
+
+    let group = bySession.get(sessionId);
+    if (!group) {
+        group = new Set();
+        bySession.set(sessionId, group);
+    }
+    group.add(client);
+}
+
+/** Remove a client from its session group, dropping the group when empty. */
+export function unindexClient(client: Client) {
+    const group = bySession.get(client.sessionId);
+    if (!group) return;
+
+    group.delete(client);
+    if (group.size === 0) bySession.delete(client.sessionId);
+}

apps/ws-server/src/index.ts

@@ -2,7 +2,7 @@ import { configDotenv } from 'dotenv';
 configDotenv();
 import WebSocket, { WebSocketServer } from 'ws';
 
-import { clients } from './clients/index.js';
+import { clients, unindexClient } from './clients/index.js';
 import { handleMessage } from './utils/handleMessages.js';
 import { startSessionSubscriber } from './utils/broadcastTosession.js';
 import { invalidateParticipants } from './utils/cache.js';
@@ -84,6 +84,7 @@ wss.on('connection', (ws: WebSocket) => {
 
         console.log('Client disconnected...', client.participantId ?? 'unknown participant');
         clients.delete(client);
+        unindexClient(client);
 
         if (client.sessionId && client.role === 'PARTICIPANT') {
             await invalidateParticipants(client.sessionId);

apps/ws-server/src/utils/broadcastTosession.ts

@@ -1,5 +1,5 @@
 import WebSocket from "ws";
-import { clients } from "../clients/index.js";
+import { clientsInSession } from "../clients/index.js";
 import { pub, sub, sessionChannel } from "../redis.js";
 import {randomUUID} from 'crypto';
 
@@ -10,9 +10,14 @@ const SERVER_ID = randomUUID();
 export function broadcastToSession(sessionId: string, type: string, payload: any = {}) {
   const envelope = JSON.stringify({ type, payload, _sid: SERVER_ID });
 
-  for(const client of clients) {
-    if(client.sessionId === sessionId && client.ws.readyState === WebSocket.OPEN) {
-      client.ws.send(JSON.stringify({ type, payload }));
+  // Serialise once for the whole session rather than per recipient: the
+  // timer ticks once a second per live quiz, so this runs on every tick for
+  // every participant.
+  const outbound = JSON.stringify({ type, payload });
+
+  for (const client of clientsInSession(sessionId)) {
+    if (client.ws.readyState === WebSocket.OPEN) {
+      client.ws.send(outbound);
     }
   }
 
@@ -36,11 +41,8 @@ export function startSessionSubscriber() {
         const sessionId = channel.replace('session:', '');
         const outbound = JSON.stringify(message);
 
-        for (const client of clients) {
-            if (
-                client.sessionId === sessionId &&
-                client.ws.readyState === WebSocket.OPEN
-            ) {
+        for (const client of clientsInSession(sessionId)) {
+            if (client.ws.readyState === WebSocket.OPEN) {
                 client.ws.send(outbound);
             }
         }

apps/ws-server/src/utils/cache.ts

@@ -102,6 +102,44 @@ export const invalidateParticipants = async(sessionId: string) => {
 }
 
 
+/**
+ * Add one participant to the cached list, returning the updated list.
+ *
+ * Joining used to invalidate the cache and immediately rebuild it, so every
+ * join re-read the whole participant table: 300 people joining one session
+ * meant ~45,000 rows read. Appending keeps the cache warm through the join
+ * rush instead.
+ *
+ * Falls back to a normal read-through when the cache is cold or unreadable,
+ * so a Redis hiccup degrades to the old behaviour rather than losing anyone.
+ */
+export const appendCachedParticipant = async (sessionId: string, participant: any) => {
+    const key = Keys.participant(sessionId);
+
+    let cached: any[] | null = null;
+    try {
+        const raw = await cache.get(key);
+        if (raw) cached = JSON.parse(raw);
+    } catch {
+        cached = null;
+    }
+
+    if (!Array.isArray(cached)) {
+        // Cold cache: read through, which now includes this participant since
+        // the row is already committed by the time we are called.
+        return getCachedParticipants(sessionId);
+    }
+
+    // A reconnect re-joins with an id already in the list.
+    if (!cached.some((p) => p?.id === participant?.id)) {
+        cached.push(participant);
+    }
+
+    await cache.setex(key, TTL.PARTICIPANT, JSON.stringify(cached));
+    return cached;
+}
+
+
 // get cached leaderboard for a session
 export async function getCachedLeaderboard(sessionId: string) {
     return readThrough(

apps/ws-server/src/utils/handleMessages.ts

@@ -3,7 +3,7 @@ import WebSocket from "ws";
 import type { Client } from "../index.js";
 import { cache } from "../redis.js";
 import { broadcastToSession } from "./broadcastTosession.js";
-import { clients, sessionTimers } from "../clients/index.js";
+import { clients, sessionTimers, indexClient, unindexClient } from "../clients/index.js";
 import { endQuizSession } from "./endQuizSession.js";
 import { getLeaderboard } from "./leaderboard.js";
 import {
@@ -12,6 +12,7 @@ import {
     getCachedParticipants,
     getCachedLeaderboard,
     invalidateParticipants,
+    appendCachedParticipant,
     invalidateLeaderboard,
     clearSessionCache,
 } from './cache.js';
@@ -102,11 +103,16 @@ export const handleMessage = async (client: Client, data: any) => {
                 for (const existing of clients) {
                     if (existing !== client && existing.participantId === participantId) {
                         clients.delete(existing);
+                        unindexClient(existing);
                         existing.ws.terminate();
                     }
                 }
             }
 
+            // Index before overwriting sessionId: indexClient uses the old
+            // value to remove the client from a previous session's group.
+            indexClient(client, sessionId);
+
             client.sessionId = sessionId;
             client.role = role;
             client.participantId = participantId;
@@ -135,28 +141,37 @@ export const handleMessage = async (client: Client, data: any) => {
                     where: { id: participantId }
                 });
 
-                await invalidateParticipants(sessionId);
-
-
-                const [participants, session] = await Promise.all([
-                    getCachedParticipants(sessionId),
-                    getCachedSession(sessionId),
-                ]);
-
-
+                const session = await getCachedSession(sessionId);
 
-                // this will also send the joincode for the participants
-                broadcastToSession(sessionId, 'participants:sync', {
-                    participants,
-                    joinCode: session?.quiz?.joinCode
-                });
+                // Keep the cache warm across the join rush instead of dropping
+                // and rebuilding it for every arrival.
+                const participants = await appendCachedParticipant(sessionId, participant);
+
+                // The full list goes to the joiner alone; everyone already
+                // connected only needs the one new arrival, which the client
+                // appends. Broadcasting the whole list to everyone made this
+                // O(n^2) - at 300 per session that is ~720MB of JSON per
+                // session just to fill the lobby.
+                if (client.ws.readyState === WebSocket.OPEN) {
+                    client.ws.send(JSON.stringify({
+                        type: 'participants:sync',
+                        payload: {
+                            participants,
+                            joinCode: session?.quiz?.joinCode
+                        }
+                    }));
+                }
 
                 broadcastToSession(sessionId, 'participant:joined', { participant });
 
                 // Replay the in-flight question to this client alone. Without
                 // it a reconnect lands on a blank screen until the organizer
                 // advances, so a brief network blip costs the whole question.
-                await sendQuizStateSnapshot(client, sessionId);
+                // Only a live quiz has anything to replay, so a first join
+                // into the lobby skips the extra queries entirely.
+                if (session?.status === 'IN_PROGRESS') {
+                    await sendQuizStateSnapshot(client, sessionId);
+                }
             }
             break;
         }
@@ -214,15 +229,10 @@ export const handleMessage = async (client: Client, data: any) => {
 
             if (!participantId) break;
 
-            // Reject a second submission for the same question. A reconnect
-            // replays the live question, so without this a participant who
-            // rejoins after answering could answer again and score twice.
-            const alreadyAnswered = await prisma.participantAnswer.findFirst({
-                where: { participantId, questionId },
-                select: { id: true },
-            });
-            if (alreadyAnswered) break;
-
+            // No pre-check for a duplicate submission: the unique constraint
+            // on (participantId, questionId) rejects one with P2002, which is
+            // handled below. Asking the database first would add a round-trip
+            // to every answer and still not settle a race.
             const questions = await getCachedQuestions(sessionId);
             const question = questions.find((q: any) => q.id === questionId);
             const answerMeta = question?.answers.find((a: any) => a.id === answerId);