diff --git a/apps/api/src/collab/handler.ts b/apps/api/src/collab/handler.ts new file mode 100644 index 0000000..239da17 --- /dev/null +++ b/apps/api/src/collab/handler.ts @@ -0,0 +1,165 @@ +import { and, eq, isNull } from "drizzle-orm" + +import { touchSession } from "../auth/session" +import { db, schema } from "../db" +import { + addRequest, + addTeacher, + getRequest, + hasTeacherOnline, + listRequests, + queueAheadOf, + removeRequest, + removeTeacher, + teacherSockets, + type CollabSocket, + type HelpRequest, +} from "./state" + +const TEACHER_ROLES = ["Teacher Admin", "Super Admin"] + +function isTeacher(ws: CollabSocket) { + return TEACHER_ROLES.includes(ws.data.adminType ?? "") +} + +/** 推给老师的列表条目。不含 socket,也不含任何代码内容 */ +function serializeRequest(request: HelpRequest) { + return { + studentId: request.studentId, + studentName: request.studentName, + className: request.className, + problemId: request.problemId, + problemTitle: request.problemTitle, + createdAt: request.createdAt, + status: request.status, + teacherName: request.teacherName ?? null, + } +} + +export function broadcastRequests() { + const payload = JSON.stringify({ + type: "requests", + list: listRequests().map(serializeRequest), + }) + for (const ws of teacherSockets()) ws.send(payload) +} + +function sendHelpStatus( + ws: CollabSocket, + status: "pending" | "active" | "cancelled" | "no_teacher", + extra: Record = {}, +) { + ws.send(JSON.stringify({ type: "help_status", status, ...extra })) +} + +export function handleCollabOpen(ws: CollabSocket) { + if (isTeacher(ws)) { + addTeacher(ws) + // 新上线的老师要立刻看到当前队列,不能等下一次变更 + ws.send( + JSON.stringify({ type: "requests", list: listRequests().map(serializeRequest) }), + ) + } +} + +export function handleCollabClose(ws: CollabSocket) { + if (isTeacher(ws)) removeTeacher(ws) + // 房间与请求的清理在 Task 3 补 +} + +export async function handleCollabMessage(ws: CollabSocket, raw: string) { + let message: { type?: unknown; problemId?: unknown; studentId?: unknown } + try { + message = JSON.parse(raw) as typeof message + } catch { + ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" })) + return + } + + // 心跳不查库,和 /ws/submissions 的处理一致 + if (message.type === "ping") { + ws.send(JSON.stringify({ type: "pong", timestamp: (message as any).timestamp })) + return + } + + // 握手时校验过一次不算数 —— 这条连接能挂几个小时 + if (!(await touchSession(ws.data.token))) { + ws.close(1008, "Session expired") + return + } + + switch (message.type) { + case "help_request": + await handleHelpRequest(ws, message.problemId) + return + case "help_cancel": + handleHelpCancel(ws) + return + default: + ws.send(JSON.stringify({ type: "error", message: "Invalid message" })) + } +} + +async function handleHelpRequest(ws: CollabSocket, problemId: unknown) { + if (typeof problemId !== "string" || !problemId) { + ws.send(JSON.stringify({ type: "error", message: "Invalid problemId" })) + return + } + if (isTeacher(ws)) { + ws.send(JSON.stringify({ type: "error", message: "教师不能发起求助" })) + return + } + if (!hasTeacherOnline()) { + sendHelpStatus(ws, "no_teacher") + return + } + + // 只认非比赛题:contest_id 为空的那条。比赛题不提供求助 + const [problem] = await db + .select({ title: schema.problem.title }) + .from(schema.problem) + .where( + and( + eq(schema.problem.displayId, problemId), + isNull(schema.problem.contestId), + ), + ) + .limit(1) + if (!problem) { + ws.send(JSON.stringify({ type: "error", message: "题目不存在或不支持求助" })) + return + } + + const existing = getRequest(ws.data.userId) + // 已经在协作中就不重复登记,否则会把正在进行的房间挤掉 + if (existing?.status === "active") return + + const [student] = await db + .select({ className: schema.user.className }) + .from(schema.user) + .where(eq(schema.user.id, ws.data.userId)) + .limit(1) + + addRequest({ + studentId: ws.data.userId, + studentName: ws.data.username ?? "", + className: student?.className ?? null, + problemId, + problemTitle: problem.title, + createdAt: Date.now(), + status: "pending", + socket: ws, + }) + sendHelpStatus(ws, "pending", { queueAhead: queueAheadOf(ws.data.userId) }) + broadcastRequests() +} + +function handleHelpCancel(ws: CollabSocket) { + const request = getRequest(ws.data.userId) + if (!request || request.status === "active") return + removeRequest(ws.data.userId) + broadcastRequests() +} + +/** Task 3 会把它换成真正的按房间转发。此刻房间还不存在,先收下不处理 */ +export function handleCollabBinary(_ws: CollabSocket, _data: Buffer | Uint8Array) {} diff --git a/apps/api/src/collab/state.ts b/apps/api/src/collab/state.ts new file mode 100644 index 0000000..3493be6 --- /dev/null +++ b/apps/api/src/collab/state.ts @@ -0,0 +1,83 @@ +/** + * 课堂求助的内存状态。 + * + * 不落库是有意的:求助是课堂上的即时行为,学生关掉页面这条请求就该消失。 + * 服务端只有一个 serve 进程(main.ts 单二进制 + 子命令,compose 里 oj-api 一个容器), + * 所以内存态够用,不需要 Redis 同步。进程重启丢掉全部状态,两端重连后回到干净状态。 + */ + +export type CollabSocket = Bun.ServerWebSocket + +export interface HelpRequest { + studentId: number + studentName: string + className: string | null + /** 题目的展示号(problem._id 列,前端一路用的都是它),不是自增主键 */ + problemId: string + problemTitle: string + createdAt: number + status: "pending" | "active" + teacherId?: number + teacherName?: string + socket: CollabSocket +} + +/** 求助表,以学生为键 —— 一个学生同时只有一个求助 */ +const requests = new Map() + +/** 在线老师的连接。用于推列表,也用于判断 no_teacher */ +const teachers = new Set() + +export function addRequest(request: HelpRequest) { + requests.set(request.studentId, request) +} + +export function getRequest(studentId: number) { + return requests.get(studentId) +} + +export function removeRequest(studentId: number) { + return requests.delete(studentId) +} + +export function hasRequest(studentId: number) { + return requests.has(studentId) +} + +/** 按发起时间正序。老师端按等待时长排序展示,不强制先来先到 */ +export function listRequests() { + return Array.from(requests.values()).sort((a, b) => a.createdAt - b.createdAt) +} + +/** 比自己早创建、且仍在排队的请求数 */ +export function queueAheadOf(studentId: number) { + const self = requests.get(studentId) + if (!self) return 0 + let ahead = 0 + for (const request of requests.values()) { + if (request.status === "pending" && request.createdAt < self.createdAt) ahead += 1 + } + return ahead +} + +export function addTeacher(ws: CollabSocket) { + teachers.add(ws) +} + +export function removeTeacher(ws: CollabSocket) { + teachers.delete(ws) +} + +export function hasTeacherOnline() { + return teachers.size > 0 +} + +export function teacherSockets() { + return teachers +} + +/** 仅供进程退出或测试用,正常路径不该调 */ +export function resetCollabState() { + requests.clear() + teachers.clear() +} diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index db78c5e..e0d0244 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -100,19 +100,30 @@ const server = Bun.serve({ new Response("Not found", { status: 404 }) ) } - if (url.pathname === "/ws/submissions" || url.pathname === "/ws/config") { + if ( + url.pathname === "/ws/submissions" || + url.pathname === "/ws/config" || + url.pathname === "/ws/collab" + ) { if (!isAllowedWebSocketOrigin(request.headers.get("origin"), url)) { return new Response("Forbidden", { status: 403 }) } const user = await getRequestSessionUser(request) if (!user) return new Response("Unauthorized", { status: 401 }) - const kind = url.pathname === "/ws/config" ? "config" : "submissions" + const kind = + url.pathname === "/ws/config" + ? "config" + : url.pathname === "/ws/collab" + ? "collab" + : "submissions" if ( bunServer.upgrade(request, { data: { userId: user.id, kind, token: readRequestSessionToken(request), + username: user.username, + adminType: user.adminType, }, }) ) { diff --git a/apps/api/src/websocket.ts b/apps/api/src/websocket.ts index 0c1768e..284a92e 100644 --- a/apps/api/src/websocket.ts +++ b/apps/api/src/websocket.ts @@ -2,6 +2,12 @@ import { flowchartUpdateSchema, submissionUpdateSchema } from "@oj2/contract" import { and, eq } from "drizzle-orm" import { touchSession } from "./auth/session" +import { + handleCollabBinary, + handleCollabClose, + handleCollabMessage, + handleCollabOpen, +} from "./collab/handler" import { config } from "./config" import { db, schema } from "./db" import { @@ -51,12 +57,17 @@ export function isAllowedWebSocketOrigin(origin: string | null, url: URL) { export interface SubmissionSocketData { userId: number - /** 同一个 Bun.serve 只能挂一个 websocket handler,用它区分两条通道 */ - kind: "submissions" | "config" + /** 同一个 Bun.serve 只能挂一个 websocket handler,用它区分通道 */ + kind: "submissions" | "config" | "collab" /** 握手时那张会话的 token,留着定期确认它还没被登出 / 过期,见 sweepSessions */ token: string /** 令牌桶,open 时初始化,见 allowMessage */ rate?: { tokens: number; updatedAt: number } + /** 以下两项只有 kind === "collab" 时才填,握手时从会话里读 */ + username?: string + adminType?: string + /** 当前所在协作房间的房主(学生)id,见 collab/handler.ts */ + roomOwnerId?: number } /** @@ -69,11 +80,24 @@ export interface SubmissionSocketData { const RATE_BURST = 20 const RATE_REFILL_PER_SECOND = 2 -function allowMessage(ws: Bun.ServerWebSocket) { +/** + * collab 通道的二进制帧(Yjs update / awareness)单独一档。 + * + * 它不查库、不解析,纯内存按房间转发,成本和文本控制帧完全不是一个量级; + * 而连续快速输入大约 5-10 帧/秒,用严格档几秒钟就会把正在协作的人踢下线。 + */ +const COLLAB_BINARY_BURST = 200 +const COLLAB_BINARY_REFILL_PER_SECOND = 100 + +function allowMessage( + ws: Bun.ServerWebSocket, + burst = RATE_BURST, + refillPerSecond = RATE_REFILL_PER_SECOND, +) { const now = Date.now() - const rate = (ws.data.rate ??= { tokens: RATE_BURST, updatedAt: now }) - const refill = ((now - rate.updatedAt) / 1000) * RATE_REFILL_PER_SECOND - rate.tokens = Math.min(RATE_BURST, rate.tokens + refill) + const rate = (ws.data.rate ??= { tokens: burst, updatedAt: now }) + const refill = ((now - rate.updatedAt) / 1000) * refillPerSecond + rate.tokens = Math.min(burst, rate.tokens + refill) rate.updatedAt = now if (rate.tokens < 1) return false rate.tokens -= 1 @@ -165,6 +189,10 @@ export function submissionWebSocketHandler(): Bun.WebSocketHandler { + console.error("Failed to handle collab message", error) + ws.send(JSON.stringify({ type: "error", message: "Internal error" })) + }) + return + } if (!allowMessage(ws)) { ws.close(1008, "Too many messages") return @@ -187,6 +234,10 @@ export function submissionWebSocketHandler(): Bun.WebSocketHandler