列表接口按行发查询是从阶段 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
f00c941ede
commit
a75c70c82d
9 files changed
+392
-247
No files matched your search
@@ -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<string, unknow
|
||||
if (onlyMeta) filters.push(eq(schema.achievement.metric, "achievement_unlocked_count"))
|
||||
else filters.push(ne(schema.achievement.metric, "achievement_unlocked_count"))
|
||||
const candidates = await db.select().from(schema.achievement).where(and(...filters))
|
||||
const created: typeof schema.achievement.$inferSelect[] = []
|
||||
for (const achievement of candidates) {
|
||||
const hits = candidates.filter((achievement) => {
|
||||
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`,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user