From a75c70c82d5f030aa71bcf2848d4a8355da2bb0b Mon Sep 17 00:00:00 2001 From: yuetsh <517252939@qq.com> Date: Wed, 26 Aug 2026 23:43:57 -0600 Subject: [PATCH] =?UTF-8?q?perf(=E5=90=8E=E7=AB=AF):=20=E5=B9=B2=E6=8E=89?= =?UTF-8?q?=2014=20=E5=A4=84=20N+1=20=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 列表接口按行发查询是从阶段 3 一路带过来的写法：`Promise.all(rows.map(...))` 看着是并发的，但每行都往库里打一次，行数一多就是几百上千次往返。同样的模式 也散在几个后台批处理和写入路径里。全部改成「先收集 id，一条 inArray/group by 查回来建 Map」——`routes/problem.ts` 的 getProblemTags 早就是这么写的，这次 只是把剩下的地方对齐。 用户可见的列表： - `GET /problem-sets` 每行 5 条（题目数/我的进度/奖章/已获奖章/创建者）， limit 上限 250 就是 1250 次往返。改成固定 5 条，与行数无关。 - `GET /contests` 每场比赛一条 creator 查询。后台的比赛列表本来就是 join 出来的，只有这条公开列表漏了。 - `GET /admin/problems`、`GET /admin/contests/:id/problems` 每题一条标签查询。 - `GET /admin/problem-sets` 每行 3 条；`.../badges` 每个奖章一条 count。 后台批处理： - `refreshContestJoinedForAll` 原来是每个用户 1 条 count 加一个独立事务里的 insert/select for update/update。改成一条 group by 出全部用户的场次，再分批 upsert，`metrics || excluded.metrics` 是 jsonb 浅合并，只覆盖 contest_joined 一个键，其余指标原样保留 —— 合并在一条语句里完成，for update 那把锁不再需要。 - `rescanAchievement` 补发循环、`unlockAchievements`：命中的一次插完， onConflictDoNothing 的 returning 就是真新解锁的那批，unlockCount 改成一次 +N。 - `resyncProgress` 逐行 UPDATE 改成一条，completed 用 least() 夹住。 写入路径： - 标签解析抽出 normalizeTagNames + findTagsByName（一条 lower(name) IN）， 新建题、改题、批量打标签三条路共用。 - 克隆比赛：题面一条 INSERT、标签一条 SELECT 加一条 INSERT。新旧题的对应 关系靠 _id 认，不依赖 returning 的行序。 - `POST /admin/website` 8 个键一条多行 upsert。 - 题单奖章判定一次插完。 `/ai/duration`：原来每个时间桶两条查询、桶之间还串行，一年 12 个桶 24 次往返。 改成先算桶、再一条查询把整段区间拉回来在内存里分桶。时间戳用 `extract(epoch) * 1000` 取毫秒回来比，别指望 Date.parse 认 pg 那个 `2026-08-12 00:00:00+00` 格式。**相邻桶首尾相接、两端闭区间**（落在边界上的 提交两个桶都算）这条旧语义是照搬的，不要顺手改成半开区间。 验证：本机起 dev 栈，造了覆盖各分支的种子数据（创建者重复的题单、零题目/零 奖章的题单、completed > total 的脏进度、除不尽的百分比、大小写混写的已有标签、 带/不带标签的比赛题、正好落在分桶边界上的提交），旧代码跑一遍、新代码跑一遍： - 51 个接口响应逐字节一致 - 12 张表的快照逐行一致（唯一差别是 progress_detail 里的 submit_time 墙钟值） - 打开 log_statement=all 数过条数，例如 `GET /admin/problems` 20 道题 24 → 5，`GET /problem-sets` 28 → 8，`/ai/duration` years:1 28 → 5， 202 个用户的成就补发 1828 → 614 一处可观察的行为变化：补发现在整批共用一个 unlockTime，原来是每人一个 new Date()。rescanAchievement 上方的注释本来就写着补发会给几百人盖同一个 时间戳、前端据此只显示「已获得」不显示日期，所以这个方向是对的。 Co-Authored-By: Claude Opus 5 --- apps/api/src/routes/admin/conf.ts | 19 +-- apps/api/src/routes/admin/contest.ts | 51 +++---- apps/api/src/routes/admin/problem.ts | 79 ++++++++--- apps/api/src/routes/admin/problemset.ts | 103 ++++++++------ apps/api/src/routes/admin/tag.ts | 30 ++--- apps/api/src/routes/ai.ts | 39 +++++- apps/api/src/routes/contest.ts | 33 +++-- apps/api/src/routes/problemset.ts | 114 ++++++++++------ apps/api/src/services/achievements.ts | 171 +++++++++++++----------- 9 files changed, 392 insertions(+), 247 deletions(-) diff --git a/apps/api/src/routes/admin/conf.ts b/apps/api/src/routes/admin/conf.ts index 1ece8d1..507f1f3 100644 --- a/apps/api/src/routes/admin/conf.ts +++ b/apps/api/src/routes/admin/conf.ts @@ -74,14 +74,17 @@ adminConfRoutes.post("/website", requireSuperAdmin, async (c) => { if (!parsed.success) { return failure(c, 400, "invalid-request", parsed.error.issues[0]?.message ?? "Invalid payload") } - for (const [field, key] of Object.entries(OPTION_KEYS) as [keyof typeof OPTION_KEYS, string][]) { - const value = parsed.data[field] - await db.insert(schema.optionsSysoptions).values({ key, value }) - .onConflictDoUpdate({ target: schema.optionsSysoptions.key, set: { value } }) - // 广播给所有开着页面的人，改完立刻生效不必刷新，对齐旧 push_config_update。 - // 推的是 options 表里的 snake_case key —— 前端 configStore.config 用的就是这套键名。 - await publishConfigUpdate(key, value) - } + const entries = (Object.entries(OPTION_KEYS) as [keyof typeof OPTION_KEYS, string][]) + .map(([field, key]) => ({ key, value: parsed.data[field] })) + // 8 个键一条 upsert 写完，不再一个键一次往返 + await db.insert(schema.optionsSysoptions).values(entries) + .onConflictDoUpdate({ + target: schema.optionsSysoptions.key, + set: { value: sql`excluded.value` }, + }) + // 广播给所有开着页面的人，改完立刻生效不必刷新，对齐旧 push_config_update。 + // 推的是 options 表里的 snake_case key —— 前端 configStore.config 用的就是这套键名。 + for (const entry of entries) await publishConfigUpdate(entry.key, entry.value) return success(c, null) }) diff --git a/apps/api/src/routes/admin/contest.ts b/apps/api/src/routes/admin/contest.ts index b1c8204..181742d 100644 --- a/apps/api/src/routes/admin/contest.ts +++ b/apps/api/src/routes/admin/contest.ts @@ -6,7 +6,7 @@ import { updateAcmHelperRequestSchema, updateContestRequestSchema, } from "@oj2/contract" -import { and, count, desc, eq, ilike } from "drizzle-orm" +import { and, count, desc, eq, ilike, inArray } from "drizzle-orm" import { Hono } from "hono" import { requireTeacher, type AppEnv } from "../../auth/middleware" @@ -206,28 +206,33 @@ adminContestRoutes.post("/contests/:id/clone", requireTeacher, async (c) => { const problems = await tx.select().from(schema.problem) .where(eq(schema.problem.contestId, id)) - for (const problem of problems) { - const { id: _oldId, ...rest } = problem - const [copy] = await tx.insert(schema.problem).values({ - ...rest, - contestId: contest!.id, - // 计数器归零：克隆的是题面，不是历史战绩 - submissionNumber: 0, - acceptedNumber: 0, - statisticInfo: {}, - createdById: me, - createTime: now, - lastUpdateTime: now, - }).returning({ id: schema.problem.id }) - // 标签是多对多中间表，Django 的 problem.tags.set(tags) 对应这里手工复制关系行 - const tags = await tx.select({ tagId: schema.problemTags.problemtagId }) - .from(schema.problemTags).where(eq(schema.problemTags.problemId, problem.id)) - if (tags.length) { - await tx.insert(schema.problemTags).values(tags.map((tag) => ({ - problemId: copy!.id, - problemtagId: tag.tagId, - }))) - } + if (problems.length === 0) return contest!.id + + // 题面、标签各一条语句，不再按题循环。新旧题的对应关系靠 _id 认： + // 克隆出来的题原样保留 _id，而它们全在同一场新比赛里，彼此不会重名。 + const copies = await tx.insert(schema.problem).values(problems.map(({ id: _oldId, ...rest }) => ({ + ...rest, + contestId: contest!.id, + // 计数器归零：克隆的是题面，不是历史战绩 + submissionNumber: 0, + acceptedNumber: 0, + statisticInfo: {}, + createdById: me, + createTime: now, + lastUpdateTime: now, + }))).returning({ id: schema.problem.id, displayId: schema.problem.displayId }) + const newIdByDisplayId = new Map(copies.map((copy) => [copy.displayId, copy.id])) + + // 标签是多对多中间表，Django 的 problem.tags.set(tags) 对应这里手工复制关系行 + const tags = await tx.select({ problemId: schema.problemTags.problemId, tagId: schema.problemTags.problemtagId }) + .from(schema.problemTags).where(inArray(schema.problemTags.problemId, problems.map((problem) => problem.id))) + if (tags.length) { + const displayIdByOldId = new Map(problems.map((problem) => [problem.id, problem.displayId])) + const links = tags.flatMap((tag) => { + const newId = newIdByDisplayId.get(displayIdByOldId.get(tag.problemId) ?? "") + return newId === undefined ? [] : [{ problemId: newId, problemtagId: tag.tagId }] + }) + if (links.length) await tx.insert(schema.problemTags).values(links) } return contest!.id }) diff --git a/apps/api/src/routes/admin/problem.ts b/apps/api/src/routes/admin/problem.ts index 25c3d59..041ffb3 100644 --- a/apps/api/src/routes/admin/problem.ts +++ b/apps/api/src/routes/admin/problem.ts @@ -54,28 +54,63 @@ async function canEdit(user: AuthUser, problem: ProblemRow) { } async function tagNames(problemId: number) { - const rows = await db.select({ name: schema.problemTag.name }).from(schema.problemTags) - .innerJoin(schema.problemTag, eq(schema.problemTags.problemtagId, schema.problemTag.id)) - .where(eq(schema.problemTags.problemId, problemId)) - return rows.map((row) => row.name) + return (await tagNamesFor([problemId])).get(problemId) ?? [] } -/** 把标签名解析成 id：去空格、大小写不敏感复用已有标签，没有才新建。对齐旧 resolve_tags */ -async function resolveTags(tx: typeof db, names: string[]) { - const ids: number[] = [] +/** 批量版：列表接口一定要走这个，按行调 tagNames 就是 N+1 */ +async function tagNamesFor(problemIds: number[]) { + const result = new Map() + if (problemIds.length === 0) return result + const rows = await db.select({ problemId: schema.problemTags.problemId, name: schema.problemTag.name }) + .from(schema.problemTags) + .innerJoin(schema.problemTag, eq(schema.problemTags.problemtagId, schema.problemTag.id)) + .where(inArray(schema.problemTags.problemId, problemIds)) + for (const row of rows) result.set(row.problemId, [...(result.get(row.problemId) ?? []), row.name]) + return result +} + +/** + * 把一批标签名去空格、去重（大小写不敏感），保留每个名字第一次出现时的原始大小写。 + * 解析和批量打标签两条路都用它，口径对齐旧 resolve_tags。 + */ +export function normalizeTagNames(names: string[]) { + const wanted: string[] = [] const seen = new Set() for (const raw of names) { const name = raw.trim() if (!name || seen.has(name.toLowerCase())) continue seen.add(name.toLowerCase()) - const [existing] = await tx.select({ id: schema.problemTag.id }).from(schema.problemTag) - .where(sql`lower(${schema.problemTag.name}) = lower(${name})`).limit(1) - if (existing) { ids.push(existing.id); continue } - const [created] = await tx.insert(schema.problemTag).values({ name }) - .returning({ id: schema.problemTag.id }) - ids.push(created!.id) + wanted.push(name) } - return ids + return wanted +} + +/** + * 一条 `lower(name) in (...)` 把已有标签全查回来，按小写名建 Map。 + * 以前是每个名字一条 SELECT，一次改十个标签就是十次往返。 + */ +export async function findTagsByName(tx: typeof db, names: string[]) { + const map = new Map() + if (names.length === 0) return map + const rows = await tx.select({ id: schema.problemTag.id, name: schema.problemTag.name }) + .from(schema.problemTag) + .where(inArray(sql`lower(${schema.problemTag.name})`, names.map((name) => name.toLowerCase()))) + for (const row of rows) map.set(row.name.toLowerCase(), row.id) + return map +} + +/** 把标签名解析成 id：去空格、大小写不敏感复用已有标签，没有才新建。对齐旧 resolve_tags */ +async function resolveTags(tx: typeof db, names: string[]) { + const wanted = normalizeTagNames(names) + if (wanted.length === 0) return [] + const existing = await findTagsByName(tx, wanted) + const missing = wanted.filter((name) => !existing.has(name.toLowerCase())) + if (missing.length) { + const created = await tx.insert(schema.problemTag).values(missing.map((name) => ({ name }))) + .returning({ id: schema.problemTag.id, name: schema.problemTag.name }) + for (const row of created) existing.set(row.name.toLowerCase(), row.id) + } + return wanted.map((name) => existing.get(name.toLowerCase())!).filter((id) => id !== undefined) } async function setTags(tx: typeof db, problemId: number, names: string[]) { @@ -254,9 +289,10 @@ adminProblemRoutes.get("/problems", requireProblemPermission, async (c) => { .where(where).orderBy(desc(schema.problem.createTime)).limit(limit).offset(offset), ]) // 只有公开题列表下发最高票评价，比赛题列表不下发 —— 与旧后端一致 - const topReactions = await getTopReactions(rows.map(({ problem }) => problem.id)) + const problemIds = rows.map(({ problem }) => problem.id) + const [topReactions, tags] = await Promise.all([getTopReactions(problemIds), tagNamesFor(problemIds)]) return success(c, adminProblemListSchema.parse({ - results: await Promise.all(rows.map(async ({ problem, user: creator, realName }) => + results: rows.map(({ problem, user: creator, realName }) => adminProblemListItemSchema.parse({ id: problem.id, _id: problem.displayId, @@ -265,12 +301,12 @@ adminProblemRoutes.get("/problems", requireProblemPermission, async (c) => { visible: problem.visible, createTime: problem.createTime, difficulty: problem.difficulty, - tags: await tagNames(problem.id), + tags: tags.get(problem.id) ?? [], hasAstRules: problem.astRules !== null, allowFlowchart: problem.allowFlowchart, showFlowchart: problem.showFlowchart, topReaction: topReactions.get(problem.id) ?? null, - }))), + })), total: totalRow[0]?.value ?? 0, })) }) @@ -425,8 +461,9 @@ adminProblemRoutes.get("/contests/:contestId/problems", requireProblemPermission .leftJoin(schema.userProfile, eq(schema.userProfile.userId, schema.user.id)) .where(where).orderBy(desc(schema.problem.createTime)).limit(limit).offset(offset), ]) + const tags = await tagNamesFor(rows.map(({ problem }) => problem.id)) return success(c, adminProblemListSchema.parse({ - results: await Promise.all(rows.map(async ({ problem, user: creator, realName }) => + results: rows.map(({ problem, user: creator, realName }) => adminProblemListItemSchema.parse({ id: problem.id, _id: problem.displayId, @@ -435,12 +472,12 @@ adminProblemRoutes.get("/contests/:contestId/problems", requireProblemPermission visible: problem.visible, createTime: problem.createTime, difficulty: problem.difficulty, - tags: await tagNames(problem.id), + tags: tags.get(problem.id) ?? [], hasAstRules: problem.astRules !== null, allowFlowchart: problem.allowFlowchart, showFlowchart: problem.showFlowchart, topReaction: null, - }))), + })), total: totalRow[0]?.value ?? 0, })) }) diff --git a/apps/api/src/routes/admin/problemset.ts b/apps/api/src/routes/admin/problemset.ts index 20484bf..4f88db8 100644 --- a/apps/api/src/routes/admin/problemset.ts +++ b/apps/api/src/routes/admin/problemset.ts @@ -42,28 +42,43 @@ async function loadOwned(c: { req: { param(name: string): string } }, user: Auth } async function serialize(row: typeof schema.problemset.$inferSelect) { - const [[problems], [participants], [creator]] = await Promise.all([ - db.select({ value: count() }).from(schema.problemsetProblem) - .where(eq(schema.problemsetProblem.problemsetId, row.id)), - db.select({ value: count() }).from(schema.problemsetProgress) - .where(eq(schema.problemsetProgress.problemsetId, row.id)), + return (await serializeMany([row]))[0]! +} + +/** 批量版：列表接口走这个，固定 3 条查询，与行数无关（按行 serialize 就是 N+1） */ +async function serializeMany(rows: (typeof schema.problemset.$inferSelect)[]) { + if (rows.length === 0) return [] + const ids = rows.map((row) => row.id) + const [problems, participants, creators] = await Promise.all([ + db.select({ problemsetId: schema.problemsetProblem.problemsetId, value: count() }) + .from(schema.problemsetProblem).where(inArray(schema.problemsetProblem.problemsetId, ids)) + .groupBy(schema.problemsetProblem.problemsetId), + db.select({ problemsetId: schema.problemsetProgress.problemsetId, value: count() }) + .from(schema.problemsetProgress).where(inArray(schema.problemsetProgress.problemsetId, ids)) + .groupBy(schema.problemsetProgress.problemsetId), db.select({ id: schema.user.id, username: schema.user.username, realName: schema.userProfile.realName }) .from(schema.user).leftJoin(schema.userProfile, eq(schema.userProfile.userId, schema.user.id)) - .where(eq(schema.user.id, row.createdById)).limit(1), + .where(inArray(schema.user.id, [...new Set(rows.map((row) => row.createdById))])), ]) - return adminProblemSetSchema.parse({ - id: row.id, - title: row.title, - description: row.description, - difficulty: row.difficulty, - status: row.status, - endTime: row.endTime, - visible: row.visible, - createdBy: sampleUser(creator ?? { id: row.createdById, username: "" }, creator?.realName), - createTime: row.createTime, - lastUpdateTime: row.lastUpdateTime, - problemsCount: problems?.value ?? 0, - participantCount: participants?.value ?? 0, + const problemsBySet = new Map(problems.map((item) => [item.problemsetId, item.value])) + const participantsBySet = new Map(participants.map((item) => [item.problemsetId, item.value])) + const creatorById = new Map(creators.map((item) => [item.id, item])) + return rows.map((row) => { + const creator = creatorById.get(row.createdById) + return adminProblemSetSchema.parse({ + id: row.id, + title: row.title, + description: row.description, + difficulty: row.difficulty, + status: row.status, + endTime: row.endTime, + visible: row.visible, + createdBy: sampleUser(creator ?? { id: row.createdById, username: "" }, creator?.realName), + createTime: row.createTime, + lastUpdateTime: row.lastUpdateTime, + problemsCount: problemsBySet.get(row.id) ?? 0, + participantCount: participantsBySet.get(row.id) ?? 0, + }) }) } @@ -97,7 +112,7 @@ adminProblemSetRoutes.get("/problem-sets", requireTeacher, async (c) => { .orderBy(desc(schema.problemset.createTime)).limit(limit).offset(offset), ]) return success(c, adminProblemSetListSchema.parse({ - results: await Promise.all(rows.map(serialize)), + results: await serializeMany(rows), total: totalRows[0]?.value ?? 0, })) }) @@ -268,9 +283,17 @@ adminProblemSetRoutes.delete("/problem-sets/:id/problems/:itemId", requireTeache // ---------------------------------------------------------------- 奖章 async function badgeWithCount(badge: BadgeRow) { - const [earned] = await db.select({ value: count() }).from(schema.userBadge) - .where(eq(schema.userBadge.badgeId, badge.id)) - return adminProblemSetBadgeSchema.parse({ + return (await badgesWithCount([badge]))[0]! +} + +/** 批量版：一条 group by 数完整批奖章的获得人数 */ +async function badgesWithCount(badges: BadgeRow[]) { + if (badges.length === 0) return [] + const earned = await db.select({ badgeId: schema.userBadge.badgeId, value: count() }) + .from(schema.userBadge).where(inArray(schema.userBadge.badgeId, badges.map((badge) => badge.id))) + .groupBy(schema.userBadge.badgeId) + const countByBadge = new Map(earned.map((item) => [item.badgeId, item.value])) + return badges.map((badge) => adminProblemSetBadgeSchema.parse({ id: badge.id, problemsetId: badge.problemsetId, name: badge.name, @@ -278,8 +301,8 @@ async function badgeWithCount(badge: BadgeRow) { icon: badge.icon, conditionType: badge.conditionType, conditionValue: badge.conditionValue, - earnedCount: earned?.value ?? 0, - }) + earnedCount: countByBadge.get(badge.id) ?? 0, + })) } /** 纯逻辑判定，对齐旧 `ProblemSetBadge._is_eligible` */ @@ -327,7 +350,7 @@ adminProblemSetRoutes.get("/problem-sets/:id/badges", requireTeacher, async (c) if (!row) return failure(c, 404, "problem-set-not-found", "题单不存在") const badges = await db.select().from(schema.problemsetBadge) .where(eq(schema.problemsetBadge.problemsetId, row.id)).orderBy(asc(schema.problemsetBadge.id)) - return success(c, await Promise.all(badges.map(badgeWithCount))) + return success(c, await badgesWithCount(badges)) }) adminProblemSetRoutes.post("/problem-sets/:id/badges", requireTeacher, async (c) => { @@ -393,21 +416,21 @@ adminProblemSetRoutes.delete("/problem-sets/:id/badges/:badgeId", requireTeacher * 只重算分母与百分比，不碰 completeTime —— 已经完成过的事实不因加题而撤销。 */ async function resyncProgress(problemsetId: number) { - const [[totalRow], progresses] = await Promise.all([ - db.select({ value: count() }).from(schema.problemsetProblem) - .where(eq(schema.problemsetProblem.problemsetId, problemsetId)), - db.select().from(schema.problemsetProgress) - .where(eq(schema.problemsetProgress.problemsetId, problemsetId)), - ]) + const [totalRow] = await db.select({ value: count() }).from(schema.problemsetProblem) + .where(eq(schema.problemsetProblem.problemsetId, problemsetId)) const total = totalRow?.value ?? 0 - for (const progress of progresses) { - const completed = Math.min(progress.completedProblemsCount, total) - await db.update(schema.problemsetProgress).set({ - totalProblemsCount: total, - completedProblemsCount: completed, - progressPercentage: total > 0 ? Math.round((completed / total) * 10000) / 100 : 0, - }).where(eq(schema.problemsetProgress.id, progress.id)) - } + // 一条 UPDATE 把整个题单的参与者刷完。以前是先把 progress 全查出来再逐行 update， + // 一个班的题单就是几十次往返，而算出来的值只跟 total 和这一行自己的 completed 有关。 + // completed 用 least(...) 夹住，百分比按 JS 那边同样的「乘 10000 四舍五入再除 100」 + // 保留两位小数 —— 数都是非负的，numeric 的 round 和 Math.round 在这个区间一致。 + const completed = sql`least(${schema.problemsetProgress.completedProblemsCount}, ${total})` + await db.update(schema.problemsetProgress).set({ + totalProblemsCount: total, + completedProblemsCount: completed, + progressPercentage: total > 0 + ? sql`round((${completed}::numeric / ${total}) * 10000) / 100` + : sql`0`, + }).where(eq(schema.problemsetProgress.problemsetId, problemsetId)) } adminProblemSetRoutes.get("/problem-sets/:id/progress", requireTeacher, async (c) => { diff --git a/apps/api/src/routes/admin/tag.ts b/apps/api/src/routes/admin/tag.ts index 5834936..59c9221 100644 --- a/apps/api/src/routes/admin/tag.ts +++ b/apps/api/src/routes/admin/tag.ts @@ -19,6 +19,7 @@ import { failure, success } from "../../http" import { JudgeStatus } from "../../judge/status" import { completeChat } from "../../services/ai" import { queryInteger, rounded } from "../helpers" +import { findTagsByName, normalizeTagNames } from "./problem" export const adminTagRoutes = new Hono() @@ -113,29 +114,20 @@ adminTagRoutes.post("/problems/batch-tag", requireProblemPermission, async (c) = if (problems.length === 0) return failure(c, 404, "no-problems", "没有可操作的题目") // 去重且大小写不敏感，与旧 resolve_tags / find_tags 一致 - const wanted: string[] = [] - const seen = new Set() - for (const raw of parsed.data.tagNames) { - const name = raw.trim() - if (!name || seen.has(name.toLowerCase())) continue - seen.add(name.toLowerCase()) - wanted.push(name) - } + const wanted = normalizeTagNames(parsed.data.tagNames) const tagIds = await db.transaction(async (tx) => { - const ids: number[] = [] - for (const name of wanted) { - const [existing] = await tx.select({ id: schema.problemTag.id }).from(schema.problemTag) - .where(sql`lower(${schema.problemTag.name}) = lower(${name})`).limit(1) - if (existing) { ids.push(existing.id); continue } - // 添加时按需新建标签，移除时只认已有标签 —— 否则「移除」会顺手造出一堆空标签 - if (parsed.data.action === "add") { - const [created] = await tx.insert(schema.problemTag).values({ name }) - .returning({ id: schema.problemTag.id }) - ids.push(created!.id) + const existing = await findTagsByName(tx as unknown as typeof db, wanted) + // 添加时按需新建标签，移除时只认已有标签 —— 否则「移除」会顺手造出一堆空标签 + if (parsed.data.action === "add") { + const missing = wanted.filter((name) => !existing.has(name.toLowerCase())) + if (missing.length) { + const created = await tx.insert(schema.problemTag).values(missing.map((name) => ({ name }))) + .returning({ id: schema.problemTag.id, name: schema.problemTag.name }) + for (const row of created) existing.set(row.name.toLowerCase(), row.id) } } - return ids + return wanted.map((name) => existing.get(name.toLowerCase())).filter((id) => id !== undefined) }) if (tagIds.length === 0) return failure(c, 404, "no-tags", "没有匹配的标签") diff --git a/apps/api/src/routes/ai.ts b/apps/api/src/routes/ai.ts index fea8f8c..16e953c 100644 --- a/apps/api/src/routes/ai.ts +++ b/apps/api/src/routes/ai.ts @@ -158,18 +158,43 @@ aiRoutes.get("/ai/duration", requireAuth, async (c) => { : duration === "months:6" ? { count: 6, unit: "months", rewind: (date: Date) => shiftMonths(date, -7), advance: (date: Date) => shiftMonths(date, 1) } : duration === "years:1" ? { count: 12, unit: "months", rewind: (date: Date) => shiftMonths(date, -13), advance: (date: Date) => shiftMonths(date, 1) } : { count: 4, unit: "weeks", rewind: (date: Date) => new Date(date.getTime() - 5 * 7 * 864e5), advance: (date: Date) => new Date(date.getTime() + 7 * 864e5) } + // 先把 count 个时间桶算出来，再一条查询把整段区间的提交拉回来在内存里分桶。 + // 以前是每个桶两条查询、桶之间还是串行的，一年 12 个桶就是 24 次往返。 + // 相邻桶首尾相接、两端都是闭区间（end_i == start_{i+1}），落在边界上的提交 + // 两个桶都算 —— 这是旧行为，照搬，不要「顺手」改成半开区间。 let cursor = config.rewind(new Date(endText)) - const data = [] + const buckets: { start: Date; end: Date }[] = [] for (let index = 0; index < config.count; index++) { const start = config.advance(cursor) - const end = config.advance(start) + buckets.push({ start, end: config.advance(start) }) cursor = start - const [submissions, solved] = await Promise.all([ - db.select({ value: count() }).from(schema.submission).where(and(eq(schema.submission.userId, user.id), gte(schema.submission.createTime, start.toISOString()), lte(schema.submission.createTime, end.toISOString()))), - db.select({ value: countDistinct(schema.submission.problemId) }).from(schema.submission).where(and(eq(schema.submission.userId, user.id), inArray(schema.submission.result, accepted), gte(schema.submission.createTime, start.toISOString()), lte(schema.submission.createTime, end.toISOString()))), - ]) - data.push(durationDataSchema.parse({ unit: config.unit, index: config.count - 1 - index, start: start.toISOString(), end: end.toISOString(), grade: solved[0]?.value ? "B" : "", problemCount: solved[0]?.value ?? 0, submissionCount: submissions[0]?.value ?? 0 })) } + // 时间戳取 epoch 毫秒回来，比较在 JS 里做，和原来在 SQL 里比 timestamptz 等价， + // 不受 pg 那个「空格分隔 + +00 偏移」字符串格式能否被 Date.parse 认的影响 + const rows = await db.select({ + time: sql`extract(epoch from ${schema.submission.createTime}) * 1000`.mapWith(Number), + problemId: schema.submission.problemId, + result: schema.submission.result, + }).from(schema.submission).where(and( + eq(schema.submission.userId, user.id), + gte(schema.submission.createTime, buckets[0]!.start.toISOString()), + lte(schema.submission.createTime, buckets.at(-1)!.end.toISOString()), + )) + const data = buckets.map((bucket, index) => { + const from = bucket.start.getTime() + const to = bucket.end.getTime() + const inRange = rows.filter((row) => row.time >= from && row.time <= to) + const solved = new Set(inRange.filter((row) => accepted.includes(row.result)).map((row) => row.problemId)).size + return durationDataSchema.parse({ + unit: config.unit, + index: config.count - 1 - index, + start: bucket.start.toISOString(), + end: bucket.end.toISOString(), + grade: solved ? "B" : "", + problemCount: solved, + submissionCount: inRange.length, + }) + }) return success(c, data) }) diff --git a/apps/api/src/routes/contest.ts b/apps/api/src/routes/contest.ts index b13ba91..f430874 100644 --- a/apps/api/src/routes/contest.ts +++ b/apps/api/src/routes/contest.ts @@ -30,14 +30,22 @@ import { objectValue, publicTemplates, queryInteger, sampleUser, stringArray } f export const contestRoutes = new Hono() -async function creator(id: number) { - const [row] = await db.select({ id: schema.user.id, username: schema.user.username, realName: schema.userProfile.realName }) +/** 一次把这批比赛的创建者全查回来，按 userId 建 Map —— 比赛列表按行查会变成 N+1 */ +async function creators(ids: number[]) { + const map = new Map>() + if (ids.length === 0) return map + const rows = await db.select({ id: schema.user.id, username: schema.user.username, realName: schema.userProfile.realName }) .from(schema.user).leftJoin(schema.userProfile, eq(schema.userProfile.userId, schema.user.id)) - .where(eq(schema.user.id, id)).limit(1) - return sampleUser(row ?? { id, username: "" }, row?.realName) + .where(inArray(schema.user.id, ids)) + for (const row of rows) map.set(row.id, sampleUser(row, row.realName)) + return map } -async function serializeContest(contest: typeof schema.contest.$inferSelect, includeNow = false) { +function serializeContest( + contest: typeof schema.contest.$inferSelect, + createdBy: ReturnType, + includeNow = false, +) { return contestSchema.parse({ id: contest.id, title: contest.title, @@ -47,7 +55,7 @@ async function serializeContest(contest: typeof schema.contest.$inferSelect, inc endTime: contest.endTime, createTime: contest.createTime, lastUpdateTime: contest.lastUpdateTime, - createdBy: await creator(contest.createdById), + createdBy, status: contestStatus(contest), contestType: contest.password ? "Password Protected" : "Public", now: includeNow ? new Date().toISOString() : undefined, @@ -72,8 +80,12 @@ contestRoutes.get("/contests", async (c) => { db.select({ value: count() }).from(schema.contest).where(where), db.select().from(schema.contest).where(where).orderBy(desc(schema.contest.startTime)).limit(limit).offset(offset), ]) + const byId = await creators([...new Set(rows.map((row) => row.createdById))]) return success(c, contestListSchema.parse({ - results: await Promise.all(rows.map((row) => serializeContest(row))), + results: rows.map((row) => serializeContest( + row, + byId.get(row.createdById) ?? sampleUser({ id: row.createdById, username: "" }, null), + )), total: totalRow[0]?.value ?? 0, })) }) @@ -81,7 +93,12 @@ contestRoutes.get("/contests", async (c) => { contestRoutes.get("/contests/:id", async (c) => { const contest = await findVisibleContest(queryInteger(c.req.param("id"), 0, { min: 1 })) if (!contest) return failure(c, 404, "contest-not-found", "Contest does not exist") - return success(c, await serializeContest(contest, true)) + const byId = await creators([contest.createdById]) + return success(c, serializeContest( + contest, + byId.get(contest.createdById) ?? sampleUser({ id: contest.createdById, username: "" }, null), + true, + )) }) contestRoutes.post("/contests/:id/access", requireAuth, async (c) => { diff --git a/apps/api/src/routes/problemset.ts b/apps/api/src/routes/problemset.ts index e16e218..5a9e995 100644 --- a/apps/api/src/routes/problemset.ts +++ b/apps/api/src/routes/problemset.ts @@ -55,11 +55,14 @@ function progressSummary(progress: typeof schema.problemsetProgress.$inferSelect } } -async function problemSetCreator(id: number) { - const [row] = await db.select({ id: schema.user.id, username: schema.user.username, realName: schema.userProfile.realName }) +async function problemSetCreators(ids: number[]) { + const map = new Map>() + if (ids.length === 0) return map + const rows = await db.select({ id: schema.user.id, username: schema.user.username, realName: schema.userProfile.realName }) .from(schema.user).leftJoin(schema.userProfile, eq(schema.userProfile.userId, schema.user.id)) - .where(eq(schema.user.id, id)).limit(1) - return sampleUser(row ?? { id, username: "" }, row?.realName) + .where(inArray(schema.user.id, ids)) + for (const row of rows) map.set(row.id, sampleUser(row, row.realName)) + return map } function badgeData(badge: typeof schema.problemsetBadge.$inferSelect, earned?: boolean) { @@ -75,35 +78,58 @@ function badgeData(badge: typeof schema.problemsetBadge.$inferSelect, earned?: b }) } -async function serializeProblemSet( - row: ProblemSetRow, +/** + * 一次把整页题单的附属数据全查回来，再在内存里按 problemsetId 分组。 + * + * 以前是每行 5 条查询（题目数 / 我的进度 / 奖章 / 已获奖章 / 创建者）， + * limit 最大 250 就是 1250 次往返。这里固定 5 条，与行数无关。 + */ +async function serializeProblemSets( + rows: ProblemSetRow[], userId?: number, includeBadges = false, ) { - const [[problemCount], [progress], badges, earnedRows] = await Promise.all([ - db.select({ value: count() }).from(schema.problemsetProblem).where(eq(schema.problemsetProblem.problemsetId, row.id)), - userId ? db.select().from(schema.problemsetProgress).where(and(eq(schema.problemsetProgress.problemsetId, row.id), eq(schema.problemsetProgress.userId, userId))).limit(1) : Promise.resolve([]), - includeBadges ? db.select().from(schema.problemsetBadge).where(eq(schema.problemsetBadge.problemsetId, row.id)) : Promise.resolve([]), + if (rows.length === 0) return [] + const ids = rows.map((row) => row.id) + const [problemCounts, progresses, badges, earnedRows, creators] = await Promise.all([ + db.select({ problemsetId: schema.problemsetProblem.problemsetId, value: count() }) + .from(schema.problemsetProblem).where(inArray(schema.problemsetProblem.problemsetId, ids)) + .groupBy(schema.problemsetProblem.problemsetId), + userId ? db.select().from(schema.problemsetProgress) + .where(and(inArray(schema.problemsetProgress.problemsetId, ids), eq(schema.problemsetProgress.userId, userId))) + : Promise.resolve([] as (typeof schema.problemsetProgress.$inferSelect)[]), + includeBadges ? db.select().from(schema.problemsetBadge) + .where(inArray(schema.problemsetBadge.problemsetId, ids)).orderBy(asc(schema.problemsetBadge.id)) + : Promise.resolve([] as (typeof schema.problemsetBadge.$inferSelect)[]), includeBadges && userId ? db.select({ id: schema.userBadge.badgeId }).from(schema.userBadge) .innerJoin(schema.problemsetBadge, eq(schema.userBadge.badgeId, schema.problemsetBadge.id)) - .where(and(eq(schema.userBadge.userId, userId), eq(schema.problemsetBadge.problemsetId, row.id))) : Promise.resolve([]), + .where(and(eq(schema.userBadge.userId, userId), inArray(schema.problemsetBadge.problemsetId, ids))) + : Promise.resolve([] as { id: number }[]), + problemSetCreators([...new Set(rows.map((row) => row.createdById))]), ]) + const countBySet = new Map(problemCounts.map((item) => [item.problemsetId, item.value])) + const progressBySet = new Map(progresses.map((item) => [item.problemsetId, item])) + const badgesBySet = new Map() + for (const badge of badges) badgesBySet.set(badge.problemsetId, [...(badgesBySet.get(badge.problemsetId) ?? []), badge]) const earned = new Set(earnedRows.map((item) => item.id)) - return problemSetSchema.parse({ - id: row.id, - title: row.title, - description: row.description, - createdBy: await problemSetCreator(row.createdById), - createTime: row.createTime, - lastUpdateTime: row.lastUpdateTime, - difficulty: row.difficulty, - status: row.status, - endTime: row.endTime, - visible: row.visible, - problemsCount: problemCount?.value ?? 0, - completedCount: progress?.completedProblemsCount ?? 0, - userProgress: progressSummary(progress), - badges: includeBadges ? badges.map((badge) => badgeData(badge, earned.has(badge.id))) : undefined, + return rows.map((row) => { + const progress = progressBySet.get(row.id) + return problemSetSchema.parse({ + id: row.id, + title: row.title, + description: row.description, + createdBy: creators.get(row.createdById) ?? sampleUser({ id: row.createdById, username: "" }, null), + createTime: row.createTime, + lastUpdateTime: row.lastUpdateTime, + difficulty: row.difficulty, + status: row.status, + endTime: row.endTime, + visible: row.visible, + problemsCount: countBySet.get(row.id) ?? 0, + completedCount: progress?.completedProblemsCount ?? 0, + userProgress: progressSummary(progress), + badges: includeBadges ? (badgesBySet.get(row.id) ?? []).map((badge) => badgeData(badge, earned.has(badge.id))) : undefined, + }) }) } @@ -123,7 +149,7 @@ problemsetRoutes.get("/problem-sets", optionalAuth, async (c) => { db.select().from(schema.problemset).where(where).orderBy(desc(schema.problemset.createTime)).limit(limit).offset(offset), ]) return success(c, problemSetListSchema.parse({ - results: await Promise.all(rows.map((row) => serializeProblemSet(row, c.get("user")?.id, true))), + results: await serializeProblemSets(rows, c.get("user")?.id, true), total: totalRows[0]?.value ?? 0, })) }) @@ -133,7 +159,8 @@ problemsetRoutes.get("/problem-sets/:id", optionalAuth, async (c) => { const [row] = await db.select().from(schema.problemset) .where(and(eq(schema.problemset.id, id), eq(schema.problemset.visible, true), ne(schema.problemset.status, "draft"))).limit(1) if (!row) return failure(c, 404, "problem-set-not-found", "题单不存在") - return success(c, await serializeProblemSet(row, c.get("user")?.id)) + const [data] = await serializeProblemSets([row], c.get("user")?.id) + return success(c, data) }) problemsetRoutes.get("/problem-sets/:id/problems", optionalAuth, async (c) => { @@ -282,22 +309,21 @@ problemsetRoutes.put("/problem-set-progress", requireAuth, async (c) => { }) } const badges = await tx.select().from(schema.problemsetBadge).where(eq(schema.problemsetBadge.problemsetId, problemSet.id)) - const earned: typeof schema.problemsetBadge.$inferSelect[] = [] - for (const badge of badges) { - const hit = badge.conditionType === "all_problems" - ? updated.totalProblemsCount > 0 && updated.completedProblemsCount === updated.totalProblemsCount - : badge.conditionType === "problem_count" - ? updated.completedProblemsCount >= badge.conditionValue - : badge.conditionType === "score" && updated.totalScore >= badge.conditionValue - if (!hit) continue - const inserted = await tx.insert(schema.userBadge).values({ - userId: user.id, - badgeId: badge.id, - earnedTime: new Date().toISOString(), - }).onConflictDoNothing({ target: [schema.userBadge.badgeId, schema.userBadge.userId] }).returning({ id: schema.userBadge.id }) - if (inserted.length) earned.push(badge) - } - return { earned } + const hits = badges.filter((badge) => badge.conditionType === "all_problems" + ? updated.totalProblemsCount > 0 && updated.completedProblemsCount === updated.totalProblemsCount + : badge.conditionType === "problem_count" + ? updated.completedProblemsCount >= badge.conditionValue + : badge.conditionType === "score" && updated.totalScore >= badge.conditionValue) + if (hits.length === 0) return { earned: [] as (typeof schema.problemsetBadge.$inferSelect)[] } + // 达标的奖章一次插完，冲突忽略后 returning 回来的就是这次真拿到的 + const inserted = await tx.insert(schema.userBadge).values(hits.map((badge) => ({ + userId: user.id, + badgeId: badge.id, + earnedTime: new Date().toISOString(), + }))).onConflictDoNothing({ target: [schema.userBadge.badgeId, schema.userBadge.userId] }) + .returning({ badgeId: schema.userBadge.badgeId }) + const insertedIds = new Set(inserted.map((row) => row.badgeId)) + return { earned: hits.filter((badge) => insertedIds.has(badge.id)) } }) if ("error" in result && result.error) { const error = result.error diff --git a/apps/api/src/services/achievements.ts b/apps/api/src/services/achievements.ts index 70fff44..3bcd4f8 100644 --- a/apps/api/src/services/achievements.ts +++ b/apps/api/src/services/achievements.ts @@ -1,4 +1,4 @@ -import { and, count, countDistinct, eq, isNotNull, isNull, ne, notInArray, sql } from "drizzle-orm" +import { and, count, countDistinct, eq, inArray, isNotNull, isNull, ne, notInArray, sql } from "drizzle-orm" import { db, schema } from "../db" import { publishAchievementNotification } from "../events" @@ -27,25 +27,27 @@ async function unlockAchievements(userId: number, metrics: Record { const value = metrics[achievement.metric] - if (typeof value !== "number") continue - const hit = achievement.operator === "gte" ? value >= achievement.threshold : value <= achievement.threshold - if (!hit) continue - const inserted = await db.insert(schema.userAchievement).values({ - userId, - achievementId: achievement.id, - unlockTime: new Date().toISOString(), - backfilled: false, - notified: false, - }).onConflictDoNothing({ target: [schema.userAchievement.achievementId, schema.userAchievement.userId] }).returning({ id: schema.userAchievement.id }) - if (inserted.length) { - await db.update(schema.achievement).set({ unlockCount: sql`${schema.achievement.unlockCount} + 1` }).where(eq(schema.achievement.id, achievement.id)) - created.push(achievement) - } - } - return created + if (typeof value !== "number") return false + return achievement.operator === "gte" ? value >= achievement.threshold : value <= achievement.threshold + }) + if (hits.length === 0) return [] + // 命中的成就一次插完，冲突忽略后 returning 回来的就是「这次真新解锁的」。 + // 一个用户对同一个成就只会解锁一次，所以每个成就都恰好 +1，一条 UPDATE 就够。 + const inserted = await db.insert(schema.userAchievement).values(hits.map((achievement) => ({ + userId, + achievementId: achievement.id, + unlockTime: new Date().toISOString(), + backfilled: false, + notified: false, + }))).onConflictDoNothing({ target: [schema.userAchievement.achievementId, schema.userAchievement.userId] }) + .returning({ achievementId: schema.userAchievement.achievementId }) + if (inserted.length === 0) return [] + const insertedIds = new Set(inserted.map((row) => row.achievementId)) + await db.update(schema.achievement).set({ unlockCount: sql`${schema.achievement.unlockCount} + 1` }) + .where(inArray(schema.achievement.id, [...insertedIds])) + return hits.filter((achievement) => insertedIds.has(achievement.id)) } export async function updateAchievementsForSubmission(submissionId: string) { @@ -161,21 +163,8 @@ export async function updateAchievementsForProblemSet(userId: number) { return [...first, ...(await unlockAchievements(userId, metrics, true))] } -/** - * 参赛场次。旧后端 `ContestJoined` 只实现了 recompute、不走 on_submission - * （比赛提交在 build_ctx 就被跳过了），所以它只在 rescan 时刷新。这里保持同样口径： - * 去重数一遍该用户有过提交的比赛数。 - * - * 注意：迁移过来时新后端**整个漏掉了这个指标**，配在 contest_joined 上的成就 - * 会永远解锁不了。补上。 - */ -async function contestJoinedCount(userId: number) { - const [row] = await db - .select({ value: countDistinct(schema.submission.contestId) }) - .from(schema.submission) - .where(and(eq(schema.submission.userId, userId), isNotNull(schema.submission.contestId))) - return row?.value ?? 0 -} +/** 一条 INSERT 里塞多少行。5 个参数一行，离 Postgres 的 65535 个参数上限还很远 */ +const USER_ACHIEVEMENT_INSERT_CHUNK = 1000 /** * 新建成就、调低阈值、或从下架改成上架之后，把已达标的存量用户补发一遍。 @@ -203,62 +192,90 @@ export async function rescanAchievement(achievementId: number) { const stats = await db.select({ userId: schema.userStat.userId, metrics: schema.userStat.metrics }) .from(schema.userStat) - let unlocked = 0 - for (const stat of stats) { - if (already.has(stat.userId)) continue + const eligible = stats.filter((stat) => { + if (already.has(stat.userId)) return false const value = objectValue(stat.metrics)[achievement.metric] - if (typeof value !== "number") continue - const hit = achievement.operator === "gte" + if (typeof value !== "number") return false + return achievement.operator === "gte" ? value >= achievement.threshold : value <= achievement.threshold - if (!hit) continue - const inserted = await db.insert(schema.userAchievement).values({ + }) + + // 达标的人分批插，冲突忽略后 returning 回来的就是真新解锁的那批 —— 以前是每人 + // 一条 insert 加一条 unlockCount+1，全站补发一次就是几千次往返。 + // 计数改成一次 +N，通知照旧逐人推（那是 Redis，不是数据库）。 + const unlockTime = new Date().toISOString() + const unlockedUserIds: number[] = [] + for (let start = 0; start < eligible.length; start += USER_ACHIEVEMENT_INSERT_CHUNK) { + const chunk = eligible.slice(start, start + USER_ACHIEVEMENT_INSERT_CHUNK) + const inserted = await db.insert(schema.userAchievement).values(chunk.map((stat) => ({ userId: stat.userId, achievementId: achievement.id, - unlockTime: new Date().toISOString(), + unlockTime, backfilled: true, notified: false, - }).onConflictDoNothing({ target: [schema.userAchievement.achievementId, schema.userAchievement.userId] }) - .returning({ id: schema.userAchievement.id }) - if (inserted.length === 0) continue - unlocked += 1 - await db.update(schema.achievement) - .set({ unlockCount: sql`${schema.achievement.unlockCount} + 1` }) - .where(eq(schema.achievement.id, achievement.id)) - await publishAchievementNotification(stat.userId, [{ - id: achievement.id, - name: achievement.name, - description: achievement.description, - icon: achievement.icon, - rarity: achievement.rarity, - kind: "achievement", - }]) + }))).onConflictDoNothing({ target: [schema.userAchievement.achievementId, schema.userAchievement.userId] }) + .returning({ userId: schema.userAchievement.userId }) + unlockedUserIds.push(...inserted.map((row) => row.userId)) } - return { scanned: stats.length, unlocked } + if (unlockedUserIds.length) { + await db.update(schema.achievement) + .set({ unlockCount: sql`${schema.achievement.unlockCount} + ${unlockedUserIds.length}` }) + .where(eq(schema.achievement.id, achievement.id)) + for (const userId of unlockedUserIds) { + await publishAchievementNotification(userId, [{ + id: achievement.id, + name: achievement.name, + description: achievement.description, + icon: achievement.icon, + rarity: achievement.rarity, + kind: "achievement", + }]) + } + } + return { scanned: stats.length, unlocked: unlockedUserIds.length } } -/** 把所有有过比赛提交的用户的 contest_joined 重算一遍，供 rescan 前置调用 */ + +/** 同上，3 个参数一行 */ +const STAT_UPSERT_CHUNK = 1000 + +/** + * 把所有有过比赛提交的用户的 contest_joined 重算一遍，供 rescan 前置调用。 + * + * 参赛场次这个指标：旧后端 `ContestJoined` 只实现了 recompute、不走 on_submission + * （比赛提交在 build_ctx 就被跳过了），所以它只在 rescan 时刷新。这里保持同样口径： + * 去重数一遍该用户有过提交的比赛数。 + * （迁移过来时新后端整个漏掉过这个指标，配在 contest_joined 上的成就永远解锁不了。） + * + * 一条 group by 出全部用户的场次，再分批 upsert —— 以前是每个用户 1 条 count + * 加一个独立事务里的 insert/select for update/update，四次往返乘以全站用户数。 + * + * `metrics || excluded.metrics` 是 jsonb 的浅合并，只覆盖 contest_joined 这一个键， + * 其余指标原样保留，语义和原来「读出来改一个键再写回去」一致，而且不需要 for update： + * 合并在一条语句里完成，并发写不会互相盖掉。 + */ async function refreshContestJoinedForAll() { const rows = await db - .selectDistinct({ userId: schema.submission.userId }) + .select({ userId: schema.submission.userId, value: countDistinct(schema.submission.contestId) }) .from(schema.submission) .where(isNotNull(schema.submission.contestId)) - for (const { userId } of rows) { - const value = await contestJoinedCount(userId) - await db.transaction(async (tx) => { - await tx.insert(schema.userStat).values({ - userId, - metrics: {}, - updateTime: new Date().toISOString(), - }).onConflictDoNothing({ target: schema.userStat.userId }) - const [stat] = await tx.select().from(schema.userStat) - .where(eq(schema.userStat.userId, userId)).for("update").limit(1) - if (!stat) return - const merged = objectValue(stat.metrics) - merged.contest_joined = value - await tx.update(schema.userStat) - .set({ metrics: merged, updateTime: new Date().toISOString() }) - .where(eq(schema.userStat.id, stat.id)) - }) + .groupBy(schema.submission.userId) + const now = new Date().toISOString() + for (let start = 0; start < rows.length; start += STAT_UPSERT_CHUNK) { + const chunk = rows.slice(start, start + STAT_UPSERT_CHUNK) + await db.insert(schema.userStat) + .values(chunk.map((row) => ({ + userId: row.userId, + metrics: { contest_joined: row.value }, + updateTime: now, + }))) + .onConflictDoUpdate({ + target: schema.userStat.userId, + set: { + metrics: sql`${schema.userStat.metrics} || excluded.metrics`, + updateTime: sql`excluded.update_time`, + }, + }) } }