From 281e354355abeb287c7c9d62d20938dc509b4aa1 Mon Sep 17 00:00:00 2001 From: Rajkumar Date: Sun, 20 Sep 2026 11:27:17 +0530 Subject: [PATCH] (fixes) reconnect logic users and also graceful-shutdown handled with no Ghostusers --- apps/frontend/src/screens/LiveQuiz.tsx | 100 +++++++++++- .../src/controllers/sessionController.ts | 41 +++-- apps/ws-server/src/index.ts | 30 +++- apps/ws-server/src/utils/endQuizSession.ts | 25 +-- apps/ws-server/src/utils/handleMessages.ts | 142 ++++++++++++++++-- apps/ws-server/src/utils/timeManager.ts | 21 ++- .../migration.sql | 18 +++ packages/db/prisma/schema.prisma | 5 + 8 files changed, 338 insertions(+), 44 deletions(-) create mode 100644 packages/db/prisma/migrations/20260920110043_unique_answer_per_participant_question/migration.sql diff --git a/apps/frontend/src/screens/LiveQuiz.tsx b/apps/frontend/src/screens/LiveQuiz.tsx index d88c624..c66a7ea 100644 --- a/apps/frontend/src/screens/LiveQuiz.tsx +++ b/apps/frontend/src/screens/LiveQuiz.tsx @@ -38,6 +38,8 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { const username = location.state?.username; const [ws, setWs] = useState(null); + const [connectionState, setConnectionState] = + useState<'connected' | 'reconnecting' | 'failed'>('connected'); const [gameState, setGameState] = useState('WAITING'); const [participants, setParticipants] = useState([]); const [currentQuestion, setCurrentQuestion] = useState(null); @@ -71,10 +73,37 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { let socket: WebSocket | null = null; let isMounted = true; - + let reconnectAttempts = 0; + let reconnectTimer: ReturnType | null = null; + // Set when we close a socket ourselves (unmount, or replacing a stale + // one) so onclose knows not to treat it as a dropped connection. + let deliberateClose = false; const wsHost = import.meta.env.VITE_WS_URL || 'ws://localhost:8080/'; + // Exponential backoff with jitter: 1s, 2s, 4s, 8s, capped at 10s. The + // jitter matters at 500 users - without it, a server restart makes every + // client reconnect in the same instant and knock it over again. + const MAX_RECONNECT_ATTEMPTS = 8; + const reconnectDelay = () => { + const base = Math.min(1000 * 2 ** reconnectAttempts, 10_000); + return base + Math.random() * 500; + }; + + const scheduleReconnect = () => { + if (!isMounted || deliberateClose) return; + + if (reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) { + setConnectionState('failed'); + return; + } + + setConnectionState('reconnecting'); + const delay = reconnectDelay(); + reconnectAttempts++; + reconnectTimer = setTimeout(initWebSocket, delay); + }; + const initWebSocket = () => { @@ -84,6 +113,9 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { socket.onopen = () => { + reconnectAttempts = 0; + setConnectionState('connected'); + if (!sessionId) { console.log("Session is not initiated. Try again...") return; @@ -130,6 +162,23 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { setGameState(GameState.QUESTION); setQuestionStartTime(Date.now()); + break; + // Sent only to a reconnecting client, to put them back on the + // question already in flight instead of a blank screen. + case 'quiz:state-snapshot': + setCurrentQuestion(payload.question); + setCorrectAnswers([]); + setTimeLeft(payload.timeLeft); + // Restoring their pick (or blocking a fresh one when the window + // has closed) is what stops a rejoin scoring the same question + // twice; the server rejects it either way. + setSelectedAnswer( + payload.alreadyAnswered ? (payload.selectedAnswerId ?? '__answered__') : null + ); + // Scoring uses Date.now() - questionStartTime, so anchor it to + // the real remaining time rather than the moment they rejoined. + setQuestionStartTime(Date.now() - (payload.question.timeLimit - payload.timeLeft) * 1000); + setGameState(payload.expired ? GameState.RESULTS : GameState.QUESTION); break; case 'quiz:timer-tick': setTimeLeft(payload.timeLeft); @@ -150,10 +199,17 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { + // onerror is always followed by onclose, so reconnect is scheduled + // there rather than in both places. socket.onerror = () => { socket?.close(); }; + socket.onclose = () => { + if (!isMounted || deliberateClose) return; + scheduleReconnect(); + }; + setWs(socket); }; @@ -163,8 +219,13 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { return () => { isMounted = false; + deliberateClose = true; clearTimeout(timeoutId); + if (reconnectTimer) clearTimeout(reconnectTimer); if (socket) { + // Drop the handler first: closing a socket fires onclose, which would + // otherwise schedule a reconnect for a screen that is going away. + socket.onclose = null; if (socket.readyState === 0) { socket.onopen = () => socket?.close(); } else { @@ -212,10 +273,43 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { } + // Rendered above every game state, so a participant always knows the live + // connection dropped rather than silently missing questions. + const ConnectionBanner = () => { + if (connectionState === 'connected') return null; + + const failed = connectionState === 'failed'; + + return ( +
+ {failed ? ( + + Connection lost.{' '} + + + ) : ( + 'Reconnecting…' + )} +
+ ); + }; + + if (gameState === GameState.WAITING) { return (
+
@@ -298,6 +392,7 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { if (gameState === GameState.STARTING) { return (
+
@@ -325,6 +420,7 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { return (
+
@@ -432,6 +528,7 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { return (
+
@@ -492,6 +589,7 @@ function LiveQuiz({ isOrganizer = false }: LiveQuizProps) { return (
+
diff --git a/apps/http-server/src/controllers/sessionController.ts b/apps/http-server/src/controllers/sessionController.ts index fb808bf..bcd6080 100644 --- a/apps/http-server/src/controllers/sessionController.ts +++ b/apps/http-server/src/controllers/sessionController.ts @@ -38,26 +38,37 @@ export const joinQuizController = async (req: Request, res: Response) => { } - // find and create a session - let session = quiz.quizSessions[0]; + // Session lookup/creation and the participant insert go together: on + // their own, a failed participant insert would leave an empty session + // behind for the quiz. + const { session, participant } = await prisma.$transaction(async (tx) => { + // Re-read inside the transaction rather than trusting the earlier + // read: at 500 simultaneous joins many requests see no session at + // once, and each would otherwise create its own. + let session = quiz.quizSessions[0] + ?? await tx.quizSession.findFirst({ + where: { quizId: quiz.id, status: { in: ['WAITING', 'IN_PROGRESS'] } }, + }); + + if (!session) { + session = await tx.quizSession.create({ + data: { + quizId: quiz.id, + status: 'WAITING' + } + }); + } - if (!session) { - session = await prisma.quizSession.create({ + // create participnats + const participant = await tx.participant.create({ data: { - quizId: quiz.id, - status: 'WAITING' + username, + sessionId: session.id, + userId: req.userId || null } }); - } - - // create participnats - const participant = await prisma.participant.create({ - data: { - username, - sessionId: session.id, - userId: req.userId || null - } + return { session, participant }; }); return res.status(200).json({ diff --git a/apps/ws-server/src/index.ts b/apps/ws-server/src/index.ts index 544f6ac..7a3313b 100644 --- a/apps/ws-server/src/index.ts +++ b/apps/ws-server/src/index.ts @@ -20,8 +20,29 @@ export interface Client { role: 'ORGANIZER' | 'PARTICIPANT'; participantId?: string; userId?: string; + isAlive: boolean; } +// A dropped connection (NAT timeout, closed laptop, proxy cutting an idle +// socket) often never fires 'close', so the client would linger in `clients` +// forever and keep inflating participant counts. Ping every 30s and drop +// anything that has not ponged since the previous round. +const HEARTBEAT_INTERVAL_MS = 30_000; + +const heartbeat = setInterval(() => { + for (const client of clients) { + if (!client.isAlive) { + client.ws.terminate(); // fires 'close' -> normal cleanup path + continue; + } + + client.isAlive = false; + client.ws.ping(); + } +}, HEARTBEAT_INTERVAL_MS); + +wss.on('close', () => clearInterval(heartbeat)); + startSessionSubscriber(); @@ -31,10 +52,17 @@ wss.on('connection', (ws: WebSocket) => { const client: Client = { ws, sessionId: '', - role: 'PARTICIPANT' + role: 'PARTICIPANT', + isAlive: true }; clients.add(client); + // Browsers answer a protocol-level ping automatically, so this needs no + // client-side support. + ws.on('pong', () => { + client.isAlive = true; + }); + ws.on('message', async (message) => { try { diff --git a/apps/ws-server/src/utils/endQuizSession.ts b/apps/ws-server/src/utils/endQuizSession.ts index c6e6a9a..64a4475 100644 --- a/apps/ws-server/src/utils/endQuizSession.ts +++ b/apps/ws-server/src/utils/endQuizSession.ts @@ -2,17 +2,22 @@ import { prisma } from "@repo/db"; import { clearSessionCache } from "./cache.js"; export async function endQuizSession(sessionId: string) { - // Mark session as completed - const session = await prisma.quizSession.update({ - where: { id: sessionId }, - data: { status: 'COMPLETED', endedAt: new Date() }, - include: { quiz: true } - }); + // Both writes go together: if the second failed on its own, the session + // would read COMPLETED while the quiz still carried a live joinCode, and + // people could keep joining a quiz that had already finished. + await prisma.$transaction(async (tx) => { + // Mark session as completed + const session = await tx.quizSession.update({ + where: { id: sessionId }, + data: { status: 'COMPLETED', endedAt: new Date() }, + include: { quiz: true } + }); - // Clear join code and set quiz to COMPLETED so no one can join again - await prisma.quiz.update({ - where: { id: session.quizId }, - data: { joinCode: null, status: 'COMPLETED' } + // Clear join code and set quiz to COMPLETED so no one can join again + await tx.quiz.update({ + where: { id: session.quizId }, + data: { joinCode: null, status: 'COMPLETED' } + }); }); await clearSessionCache(sessionId); diff --git a/apps/ws-server/src/utils/handleMessages.ts b/apps/ws-server/src/utils/handleMessages.ts index 3d0a4da..ba45823 100644 --- a/apps/ws-server/src/utils/handleMessages.ts +++ b/apps/ws-server/src/utils/handleMessages.ts @@ -1,7 +1,9 @@ import { prisma } from "@repo/db"; +import WebSocket from "ws"; import type { Client } from "../index.js"; +import { cache } from "../redis.js"; import { broadcastToSession } from "./broadcastTosession.js"; -import { sessionTimers } from "../clients/index.js"; +import { clients, sessionTimers } from "../clients/index.js"; import { endQuizSession } from "./endQuizSession.js"; import { getLeaderboard } from "./leaderboard.js"; import { @@ -13,9 +15,70 @@ import { invalidateLeaderboard, clearSessionCache, } from './cache.js'; -import { startQuestionTimer } from "./timeManager.js"; - - +import { startQuestionTimer, timerDeadlineKey } from "./timeManager.js"; + + + + +/** + * Bring a reconnecting participant back to the live question. + * + * Sent to the one client rather than broadcast: everyone else is already in + * the right state, and re-broadcasting a question would reset their timers. + */ +async function sendQuizStateSnapshot(client: Client, sessionId: string) { + const session = await getCachedSession(sessionId); + + // Nothing in flight before the quiz starts or once it is over: the normal + // WAITING / ENDED handling already covers those. + if (!session || session.status !== 'IN_PROGRESS') return; + + const questionIndex = session.currentQuestionIndex ?? 0; + const questions = await getCachedQuestions(sessionId); + const question = questions[questionIndex]; + if (!question) return; + + // How long is actually left, from the shared deadline rather than from a + // countdown living in one instance's memory. + const rawDeadline = await cache.get(timerDeadlineKey(sessionId)); + const timeLeft = rawDeadline + ? Math.max(0, Math.ceil((Number(rawDeadline) - Date.now()) / 1_000)) + : 0; + + // Did this participant already answer? There is no unique constraint on + // (participantId, questionId), so a replayed question they could answer + // twice would score twice. + const existingAnswer = client.participantId + ? await prisma.participantAnswer.findFirst({ + where: { participantId: client.participantId, questionId: question.id }, + select: { answerId: true }, + }) + : null; + + const sanitizedQuestion = { + ...question, + answers: question.answers.map((ans: any) => ({ + id: ans.id, + text: ans.text, + })), + }; + + if (client.ws.readyState !== WebSocket.OPEN) return; + + client.ws.send(JSON.stringify({ + type: 'quiz:state-snapshot', + payload: { + question: sanitizedQuestion, + questionIndex, + timeLeft, + // Already answered, or the window closed while they were away: + // the client shows the question read-only in both cases. + alreadyAnswered: !!existingAnswer, + selectedAnswerId: existingAnswer?.answerId ?? null, + expired: timeLeft <= 0, + }, + })); +} export const handleMessage = async (client: Client, data: any) => { @@ -27,6 +90,23 @@ export const handleMessage = async (client: Client, data: any) => { console.log("JOIN EVENT") const { sessionId, role, participantId, userId } = payload; + + // A reconnecting client re-joins with the same participantId while + // its previous socket may still be registered: the heartbeat can + // take up to two rounds to notice a dead peer, and a background + // tab that never fired 'close' lingers even longer. Two entries + // for one person means every broadcast is delivered twice and the + // participant count is inflated, so retire the stale socket now + // rather than waiting for the heartbeat. + if (participantId) { + for (const existing of clients) { + if (existing !== client && existing.participantId === participantId) { + clients.delete(existing); + existing.ws.terminate(); + } + } + } + client.sessionId = sessionId; client.role = role; client.participantId = participantId; @@ -72,6 +152,11 @@ export const handleMessage = async (client: Client, data: any) => { }); 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); } break; } @@ -129,6 +214,15 @@ 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; + const questions = await getCachedQuestions(sessionId); const question = questions.find((q: any) => q.id === questionId); const answerMeta = question?.answers.find((a: any) => a.id === answerId); @@ -136,15 +230,26 @@ export const handleMessage = async (client: Client, data: any) => { const isCorrect = !!answerMeta?.isCorrect; const points = isCorrect ? Math.max(10, 1_000 - timeMs) : 0; - await Promise.all([ - prisma.participantAnswer.create({ - data: { participantId, questionId, answerId, timeMs, isCorrect, points }, - }), - prisma.participant.update({ - where: { id: participantId }, - data: { score: { increment: points } }, - }), - ]); + // Insert the answer first and only credit the score if it was + // actually recorded. Running both concurrently would award points + // even when the unique constraint rejected the answer, which is + // exactly the double-scoring this is meant to prevent. + try { + await prisma.$transaction([ + prisma.participantAnswer.create({ + data: { participantId, questionId, answerId, timeMs, isCorrect, points }, + }), + prisma.participant.update({ + where: { id: participantId }, + data: { score: { increment: points } }, + }), + ]); + } catch (err: any) { + // P2002 = unique constraint: a duplicate submission that raced + // past the check above. Nothing was written; not an error. + if (err?.code === 'P2002') break; + throw err; + } await invalidateLeaderboard(sessionId); @@ -154,17 +259,24 @@ export const handleMessage = async (client: Client, data: any) => { case 'quiz:end': { - const [sessionId] = payload; + const { sessionId } = payload; + + if (!sessionId) break; if (sessionTimers.has(sessionId)) { clearInterval(sessionTimers.get(sessionId)); sessionTimers.delete(sessionId); } + // Read the leaderboard before the caches are dropped: both + // endQuizSession and clearSessionCache clear it, and reading + // afterwards would re-query the DB and repopulate what we just + // cleared. + const leaderboard = await getCachedLeaderboard(sessionId); + await endQuizSession(sessionId); await clearSessionCache(sessionId); - const leaderboard = await getCachedLeaderboard(sessionId); broadcastToSession(sessionId, 'quiz:leaderboard', { leaderboard }); broadcastToSession(sessionId, 'quiz:ended'); break; diff --git a/apps/ws-server/src/utils/timeManager.ts b/apps/ws-server/src/utils/timeManager.ts index edda4bf..631fd07 100644 --- a/apps/ws-server/src/utils/timeManager.ts +++ b/apps/ws-server/src/utils/timeManager.ts @@ -10,6 +10,13 @@ import { sessionTimers } from '../clients/index.js'; const timerLockKey = (sessionId: string, questionIndex: number) => `quiz:timer-lock:${sessionId}:${questionIndex}`; +// Absolute epoch-ms deadline for the question currently in flight. The +// countdown itself lives in a setInterval closure on one instance, which a +// reconnecting client cannot read — this key is what lets any instance answer +// "how much time is left?" during a mid-question rejoin. +export const timerDeadlineKey = (sessionId: string) => + `quiz:deadline:${sessionId}`; + export async function startQuestionTimer( sessionId: string, questionIndex: number, @@ -43,6 +50,14 @@ export async function startQuestionTimer( } let timeLeft = question.timeLimit; + + // Published so a mid-question rejoin can compute the real remaining time. + await cache.set( + timerDeadlineKey(sessionId), + String(Date.now() + question.timeLimit * 1_000), + 'EX', question.timeLimit + 10, + ); + broadcastToSession(sessionId, 'quiz:timer-tick', { timeLeft }); const timer = setInterval(async () => { @@ -53,8 +68,10 @@ export async function startQuestionTimer( clearInterval(timer); sessionTimers.delete(sessionId); - // Release the lock immediately so a new question can start - await cache.del(lockKey); + // Release the lock immediately so a new question can start. + // The deadline goes with it: the question is over, so a rejoin + // from here on should see results, not a live countdown. + await cache.del(lockKey, timerDeadlineKey(sessionId)); await invalidateLeaderboard(sessionId); const leaderboard = await getCachedLeaderboard(sessionId); diff --git a/packages/db/prisma/migrations/20260920110043_unique_answer_per_participant_question/migration.sql b/packages/db/prisma/migrations/20260920110043_unique_answer_per_participant_question/migration.sql new file mode 100644 index 0000000..d463d34 --- /dev/null +++ b/packages/db/prisma/migrations/20260920110043_unique_answer_per_participant_question/migration.sql @@ -0,0 +1,18 @@ +-- One answer per participant per question. +-- +-- A reconnecting participant is replayed the live question, so without this +-- they could submit a second time and be scored twice. The application also +-- checks, but two submissions landing in the same instant can both pass that +-- check; only the database can settle the race. + +-- Collapse any pre-existing duplicates first, keeping the earliest answer, +-- which is the one the participant actually gave first. +DELETE FROM "ParticipantAnswer" a +USING "ParticipantAnswer" b +WHERE a."participantId" = b."participantId" + AND a."questionId" = b."questionId" + AND (a."createdAt" > b."createdAt" + OR (a."createdAt" = b."createdAt" AND a."id" > b."id")); + +CREATE UNIQUE INDEX "ParticipantAnswer_participantId_questionId_key" + ON "ParticipantAnswer" ("participantId", "questionId"); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index fa9c63e..a9780a9 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -146,4 +146,9 @@ model ParticipantAnswer { answerId String? answer Answer? @relation(fields: [answerId], references: [id]) + + // One answer per participant per question. A reconnect replays the live + // question, so this is what makes double-scoring impossible even if two + // submissions race past the application-level check. + @@unique([participantId, questionId]) }