feat(api): 新增 /ws/collab 通道与课堂求助控制面

学生发起/撤销求助，在线教师收到全量列表。求助只在内存，不落库。
二进制帧的限流单独一档，避免协作输入把连接踢掉。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K1d8B3f4SXJwDvUY625eQd

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<string, unknown> = {},
+) {
+  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) {}

apps/api/src/collab/state.ts

@@ -0,0 +1,83 @@
+/**
+ * 课堂求助的内存状态。
+ *
+ * 不落库是有意的：求助是课堂上的即时行为，学生关掉页面这条请求就该消失。
+ * 服务端只有一个 serve 进程（main.ts 单二进制 + 子命令，compose 里 oj-api 一个容器），
+ * 所以内存态够用，不需要 Redis 同步。进程重启丢掉全部状态，两端重连后回到干净状态。
+ */
+
+export type CollabSocket = Bun.ServerWebSocket<import("../websocket").SubmissionSocketData>
+
+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<number, HelpRequest>()
+
+/** 在线老师的连接。用于推列表，也用于判断 no_teacher */
+const teachers = new Set<CollabSocket>()
+
+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()
+}

apps/api/src/index.ts

@@ -100,19 +100,30 @@ const server = Bun.serve<SubmissionSocketData>({
 				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,
 					},
 				})
 			) {

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，用它区分两条通道 */
+  /** 同一个 Bun.serve 只能挂一个 websocket handler，用它区分通道 */
-  kind: "submissions" | "config"
+  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<SubmissionSocketData>) {
+/**
+ * collab 通道的二进制帧（Yjs update / awareness）单独一档。
+ *
+ * 它不查库、不解析，纯内存按房间转发，成本和文本控制帧完全不是一个量级；
+ * 而连续快速输入大约 5-10 帧/秒，用严格档几秒钟就会把正在协作的人踢下线。
+ */
+const COLLAB_BINARY_BURST = 200
+const COLLAB_BINARY_REFILL_PER_SECOND = 100
+
+function allowMessage(
+  ws: Bun.ServerWebSocket<SubmissionSocketData>,
+  burst = RATE_BURST,
+  refillPerSecond = RATE_REFILL_PER_SECOND,
+) {
   const now = Date.now()
-  const rate = (ws.data.rate ??= { tokens: RATE_BURST, updatedAt: now })
+  const rate = (ws.data.rate ??= { tokens: burst, updatedAt: now })
-  const refill = ((now - rate.updatedAt) / 1000) * RATE_REFILL_PER_SECOND
+  const refill = ((now - rate.updatedAt) / 1000) * refillPerSecond
-  rate.tokens = Math.min(RATE_BURST, rate.tokens + refill)
+  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<SubmissionSoc
     open(ws) {
       liveSockets.add(ws)
       ws.data.rate = { tokens: RATE_BURST, updatedAt: Date.now() }
+      if (ws.data.kind === "collab") {
+        handleCollabOpen(ws)
+        return
+      }
       if (ws.data.kind === "config") {
         ws.subscribe(configTopic)
         return
@@ -173,6 +201,25 @@ export function submissionWebSocketHandler(): Bun.WebSocketHandler<SubmissionSoc
       ws.subscribe(userEventTopic(ws.data.userId))
     },
     message(ws, message) {
+      if (ws.data.kind === "collab") {
+        if (typeof message !== "string") {
+          if (!allowMessage(ws, COLLAB_BINARY_BURST, COLLAB_BINARY_REFILL_PER_SECOND)) {
+            ws.close(1008, "Too many messages")
+            return
+          }
+          handleCollabBinary(ws, message)
+          return
+        }
+        if (!allowMessage(ws)) {
+          ws.close(1008, "Too many messages")
+          return
+        }
+        handleCollabMessage(ws, message).catch((error) => {
+          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<SubmissionSoc
     },
     close(ws) {
       liveSockets.delete(ws)
+      if (ws.data.kind === "collab") {
+        handleCollabClose(ws)
+        return
+      }
       if (ws.data.kind === "config") {
         ws.unsubscribe(configTopic)
         return