import * as PersonalPeriods from "@/cloudRise/periods"; import * as ConfigStore from "@/activityConfig/store"; import cloud from "@lafjs/cloud"; import Utils from "@/Utils"; import * as RuleModule from "@/cloudRise/rules"; import * as StatsModule from "@/cloudRise/stats"; import * as Analytics from "@/cloudRise/analytics"; import { createHash, randomInt } from "crypto"; const Rules: any = RuleModule; const Stats: any = StatsModule; const mongo: any = cloud.mongo.db; const users = () => mongo.collection("users"); const random = () => randomInt(0, 0x100000000) / 0x100000000; const key = (...parts: string[]) => createHash("sha256").update(JSON.stringify(parts)).digest("hex"); const fail = (msg: string) => ({ code: 0, data: null, msg }); // Utils.POST sends form fields as strings; reject non-integer text and coercible JSON values. const parseInteger = (value: any): number => typeof value === "number" ? value : typeof value === "string" && /^\d+$/.test(value) ? Number(value) : NaN; /** New clients send completed mainline levels; older queued requests keep their original fallback. */ function matchingStartLevel(body: any, user: any): number { const completed = body.levelAmount === undefined ? Number(user.levelAmount || 0) : parseInteger(body.levelAmount); if (!Number.isSafeInteger(completed) || completed < 0 || !Number.isSafeInteger(completed + 1)) { throw new Error("当前关卡数无效"); } return completed + 1; } function parseState(raw: any): any { if (raw === undefined || raw === null || raw === "") return Rules.initialState(); const state = JSON.parse(typeof raw === "string" ? raw : JSON.stringify(raw)); if (state.version !== 1 || !Array.isArray(state.playedPeriods)) throw new Error("百人赛存档版本不支持"); // Runs created before configurable pools keep their original stage amounts and defaults. if (state.run && !state.run.pools) { state.run.pools = Rules.POOLS.map((pool: number, i: number) => state.run.stages.find((stage: any) => stage.stage === i + 1)?.pool ?? pool); } return state; } function oldStateQuery(user: any): any { return { onlyId: user.onlyId, cloudRisePeriod: user.cloudRisePeriod === undefined ? { $exists: false } : user.cloudRisePeriod, cloudRiseState: user.cloudRiseState === undefined ? { $exists: false } : user.cloudRiseState }; } export function periodPools(config: any): number[] { return [...(config.pools ?? Rules.POOLS)]; } export function periodDurationHours(config: any): number { return config.durationHours === undefined ? 24 : config.durationHours; } export function validatePeriod(config: any): string | null { if (!config || typeof config.periodId !== "string" || !config.periodId.trim() || config.periodId.length > 100) return "活动期标识无效"; if (!Number.isSafeInteger(config.startsAt) || !Number.isSafeInteger(config.endsAt) || config.startsAt >= config.endsAt) return "报名起止时间无效"; if (!Number.isSafeInteger(config.unlockLevel) || config.unlockLevel < 1) return "请配置入口解锁关卡"; const hours = periodDurationHours(config); if (!PersonalPeriods.validDurationHours(hours) || !Number.isSafeInteger(config.endsAt + hours * 60 * 60 * 1000)) return "durationHours 必须为正整数小时;测试环境可用正数小数小时,且必须能表示为整数毫秒、截止时间不能越界"; if (config.pools !== undefined && (!Array.isArray(config.pools) || config.pools.length !== 3 || ![0, 1, 2].every(i => Number.isSafeInteger(config.pools[i]) && config.pools[i] > 0))) return "pools 必须为三个正安全整数"; return null; } /** Shared immutable versions; publication time orders versions without restricting enrollment. */ export async function findOpenPeriod(now: number): Promise { const rows = await ConfigStore.records().find({ activityId: "cloudRise", recordType: "version", status: "published", effectiveFrom: { $lte: now }, "config.startsAt": { $lte: now }, "config.endsAt": { $gt: now } }) .sort({ "config.startsAt": -1, "config.periodId": 1, publishedAt: -1, _id: 1 }).toArray(); const row = rows.find((r: any) => r.effectiveFrom === r.config.startsAt); return row ? { ...row.config, configVersion: row.configVersion, enabled: row.enabled !== false } : null; } function canJoin(config: any, state: any, now: number): boolean { return !!config && !validatePeriod(config) && now >= config.startsAt && now < config.endsAt && !Rules.active(state.run) && !state.playedPeriods.includes(config.periodId) && !(state.run?.stages || []).some((stage: any) => stage.status === "won" && !stage.rewardSaved); } /** Two calendar months, clamping month-end instead of overflowing into the next month. */ export function inactiveBefore(now: number): number { const date = new Date(now), day = date.getUTCDate(); date.setUTCDate(1); date.setUTCMonth(date.getUTCMonth() - 2); const last = new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth() + 1, 0)).getUTCDate(); date.setUTCDate(Math.min(day, last)); return date.getTime(); } async function makeStage(stage: number, start: number, onlyId: number, now: number, rewardPool: number): Promise { // Frozen inactive-user snapshot; no live users scan during matching. // Keep account identifiers internal and still exclude the current player. const pool = Stats.OPPONENT_PROFILES.filter((p: any) => p.onlyId !== onlyId).map((p: any) => ({ username: String(p.username || "玩家"), useravatar: typeof p.useravatar === "string" ? p.useravatar : "", useravatarIcon: String(p.useravatarIcon || "icon_0") })); // Empty pools fail before committing participation. Do not invent real users or change scores. if (!pool.length) throw new Error("暂无符合条件的参赛资料,请稍后再试"); const histogram = await mongo.collection("cloudRiseSamples").aggregate(sampleHistogramPipeline(stage)).toArray(); const samples: Record = {}; for (const item of histogram) { const startKey = String(item._id.start); if (!samples[startKey]) samples[startKey] = Array(Rules.TARGETS[stage - 1] + 1).fill(0); samples[startKey][item._id.score] = item.count; } return Rules.matchStage(stage, start, now, Stats.LEVEL_STATS, samples, pool, random, rewardPool); } /** One document per player-period; each completed stage contributes one score. */ export function sampleHistogramPipeline(stage: number): any[] { const filter = { stage, start_level: { $gte: 1, $lte: Rules.MAX_START_LEVEL }, success_num: { $gte: 0, $lte: Rules.TARGETS[stage - 1] }, outcome: { $in: ["won", "lost"] }, }; return [ { $match: { schemaVersion: 2, samples: { $elemMatch: filter } } }, { $unwind: "$samples" }, { $match: Object.fromEntries(Object.entries(filter).map(([field, value]) => ["samples." + field, value])) }, { $group: { _id: { start: "$samples.start_level", score: "$samples.success_num" }, count: { $sum: 1 } } }, ]; } /** One small document per run (uid + period); completed stages are immutable samples. */ async function persistSamples(uid: string, runId: string, periodId: string, samples: any[]): Promise { if (!samples.length) return; const collection = mongo.collection("cloudRiseSamples"); await collection.updateOne({ _id: runId }, { $setOnInsert: { schemaVersion: 2, uid, runId, periodId, samples: [] }, }, { upsert: true }); for (const sample of samples) { // Conditional append is atomic; concurrent retries and older snapshots cannot add a stage twice or remove a newer stage. await collection.updateOne({ _id: runId, "samples.stage": { $ne: sample.stage } }, { $push: { samples: sample }, }); } } /** Existing terminal archives can be compacted without touching live state or reward receipts. */ export async function compactArchives(limit: number, afterId = ""): Promise { const collection = mongo.collection("cloudRiseRuns"); const filter = { status: { $in: ["completed", "failed", "expired"] }, stages: { $elemMatch: { opponents: { $exists: true } } }, }; const query: any = { ...filter }; if (afterId) query._id = { $gt: afterId }; const rows = await collection.find(query).sort({ _id: 1 }).limit(limit).toArray(); let compacted = 0; for (const row of rows) { if (typeof row._id !== "string") throw new Error("归档标识格式无效,未清理: " + row._id); const result = await collection.updateOne({ ...filter, _id: row._id }, { $unset: { "stages.$[].opponents": "" }, }); compacted += result.modifiedCount; } return { compacted, nextCursor: rows.length === limit ? rows[rows.length - 1]._id : null }; } /** The authoritative outbox is the user's saved stage records, retained until this succeeds. */ async function persistHistory(uid: string, run: any): Promise { if (!run) return; await persistSamples(uid, run.id, run.periodId, run.stages .filter((s: any) => s.status === "won" || s.status === "lost") .map((s: any) => ({ stage: s.stage, start_level: s.start_level, success_num: s.success_num, outcome: s.status, reason: s.reason || "win", endedAt: s.endedAt, }))); if (!Rules.active(run) && run.stages.every((stage: any) => stage.status !== "won" || stage.rewardSaved)) { await mongo.collection("cloudRiseRuns").updateOne({ _id: run.id }, { $setOnInsert: { uid, ...run, // Preserve settlement/audit fields; opponent snapshots remain only in the user's own stages. stages: run.stages.map(({ opponents, ...summary }: any) => summary), } }, { upsert: true }); } } function response(user: any, state: any, config: any, now: number, sampleSyncPending = false): any { const run = state.run; // Keep durable history for retries, but do not present a settled older run as the new event. const settledPreviousPeriod = config && run && run.periodId !== config.periodId && ["completed", "failed", "expired"].includes(run.status) && run.stages.every((stage: any) => stage.status !== "won" || stage.rewardSaved); return { code: 1, msg: "ok", data: { serverNow: now, available: canJoin(config, state, now), period: config && !validatePeriod(config) ? { periodId: config.periodId, durationHours: periodDurationHours(config), startsAt: config.startsAt, endsAt: config.endsAt, unlockLevel: config.unlockLevel, pools: periodPools(config) } : null, ...(config || run ? { targets: Rules.TARGETS, pools: run && !settledPreviousPeriod ? run.pools : periodPools(config) } : {}), run: settledPreviousPeriod ? null : Rules.publicRun(run), matching: state.pendingMatch ? { id: state.pendingMatch.id, stage: Rules.publicRun(state.pendingMatch.run).stages.at(-1) } : null, ...(settledPreviousPeriod ? { previousRun: Rules.publicRun(run) } : {}), coinAmount: Number(user.coinAmount) || 0, sampleSyncPending, } }; } export default async function (ctx: FunctionContext) { const body = ctx.body || {}, action = body.action || "status"; const onlyId = typeof body.uid === "number" ? body.uid : typeof body.uid === "string" && /^[1-9]\d*$/.test(body.uid) ? Number(body.uid) : NaN; if (!Number.isSafeInteger(onlyId) || onlyId < 1) return fail("玩家 uid 必须为 users.onlyId"); const uid = String(onlyId); if (body.gameName === "iaa") return fail("当前账号类型不支持百人赛"); if (!["open_period", "status", "start", "start_stage", "begin", "finish", "save_reward", "prepare_match", "cancel_match", "confirm_match"].includes(action)) return fail("无效的活动操作"); const matchId = body.matchId; if ((matchId !== undefined || ["prepare_match", "cancel_match", "confirm_match"].includes(action)) && (typeof matchId !== "string" || !/^[A-Za-z0-9_-]{8,100}$/.test(matchId))) return fail("匹配标识无效"); try { // Retry only optimistic concurrency conflicts, always recomputing from committed state. for (let retry = 0; retry < 5; retry++) { const user = await users().findOne({ onlyId }); if (!user) return fail("玩家不存在"); if (!user.token || !Utils.checkToken(body.token, user.token)) return fail("token校验失败"); if (action === "open_period") { const period = await PersonalPeriods.openForUser(onlyId); return period ? { code: 1, msg: "ok", data: { serverNow: Date.now(), period } } : fail("尚未满足活动开启条件"); } const state = parseState(user.cloudRiseState), before = JSON.stringify(state), now = Date.now(); const candidate = PersonalPeriods.periodForUser(user); const config = candidate && !validatePeriod(candidate) && now >= candidate.startsAt && now < candidate.endsAt ? candidate : null; // Personal runs still need recovery, settlement and idempotent retries after registration closes. if (!config && !state.run && action !== "cancel_match") return fail("活动未开启"); const expired = Rules.expire(state.run, now); let reportedBalance: number | undefined; if (state.pendingMatch && now >= state.pendingMatch.validUntil) delete state.pendingMatch; // Older drafts copied all prior stages. Keep their already matched final stage, without rerolling. // Run this after capturing 'before' so status also persists the compacted draft through CAS. if (state.pendingMatch?.run.stages.length > 1) { state.pendingMatch.run.stages = [state.pendingMatch.run.stages.at(-1)]; } if (action === "cancel_match") { // Tombstones stop an in-flight preparation from restoring a cancelled draft after a CAS retry. state.cancelledMatches = [...new Set([...(state.cancelledMatches || []), matchId])].slice(-128); if (state.pendingMatch?.id === matchId) delete state.pendingMatch; } else if (action === "confirm_match") { const pending = state.pendingMatch; if (state.run?.stages.some((s: any) => s.matchId === matchId)) { // Confirmation response may have been lost; do not start again or rewrite progress. } else { if (!pending || pending.id !== matchId || state.cancelledMatches?.includes(matchId)) return fail("匹配已取消或失效,请重新开始"); if (pending.baseRunId !== (state.run?.id || null) || pending.baseStage !== (state.run?.stage || 0)) return fail("活动进度已变化,请重新匹配"); if (pending.run.stage === 1) { if (!canJoin(config, state, now) || pending.run.periodId !== config.periodId) return fail("报名已结束"); await persistHistory(uid, state.run); pending.run.startedAt = now; pending.run.expiresAt = pending.validUntil; state.playedPeriods.push(pending.run.periodId); } else { if (expired || state.run.status !== "waiting") return fail("下一阶段尚未解锁或挑战已结束"); // A reward receipt may arrive while matching; retain the latest settled stages. pending.run = { ...state.run, status: "playing", stage: pending.run.stage, stages: [...state.run.stages, pending.run.stages.at(-1)] }; } pending.run.stages.at(-1).matchId = matchId; state.run = pending.run; delete state.pendingMatch; } } else if (action === "prepare_match") { if (!state.cancelledMatches?.includes(matchId) && state.pendingMatch?.id !== matchId) { let draft: any; if (Number(body.stage) === 1) { if (!canJoin(config, state, now) || body.periodId !== config.periodId) return fail("本期不能再次参加或报名已结束"); const pools = periodPools(config), durationHours = periodDurationHours(config); draft = { id: key(uid, config.periodId), periodId: config.periodId, status: "playing", stage: 1, startedAt: now, expiresAt: config.endsAt, durationHours, pools, stages: [await makeStage(1, matchingStartLevel(body, user), onlyId, now, pools[0])] }; } else { const run = state.run, stage = Number(body.stage); if (!run || expired || run.id !== body.runId || run.status !== "waiting" || stage !== run.stage + 1 || stage > 3) return fail("下一阶段尚未解锁"); // Only the new stage belongs to the draft. Confirmation merges it into the latest saved run. draft = { ...run, stage, status: "playing", stages: [await makeStage(stage, matchingStartLevel(body, user), onlyId, now, run.pools[stage - 1])] }; } state.pendingMatch = { id: matchId, baseRunId: state.run?.id || null, baseStage: state.run?.stage || 0, validUntil: Number(body.stage) === 1 ? config.endsAt : state.run.expiresAt, run: draft }; } } else if (action === "start") { if (Rules.active(state.run)) { if (body.periodId !== state.run.periodId) return fail("已有进行中的百人赛"); } else { if (!canJoin(config, state, now) || body.periodId !== config.periodId) return fail("本期不能再次参加或报名已结束"); // Archive all previous samples before replacing their durable source records. await persistHistory(uid, state.run); const start = matchingStartLevel(body, user); const pools = periodPools(config), durationHours = periodDurationHours(config); const stage = await makeStage(1, start, onlyId, now, pools[0]); state.run = { id: key(uid, config.periodId), periodId: config.periodId, status: "playing", stage: 1, startedAt: now, expiresAt: config.endsAt, durationHours, pools, stages: [stage] }; state.playedPeriods.push(config.periodId); } } else if (action === "save_reward" || !expired && action !== "status") { const run = state.run; if (!run || body.runId !== run.id) return fail("活动轮次不匹配,请刷新"); const requestedStage = Number(body.stage); if (action === "save_reward") { const stage = run.stages.find((s: any) => s.stage === requestedStage); if (!stage || stage.status !== "won") return fail("当前阶段尚未成功"); if (!stage.rewardSaved) { const reward = parseInteger(body.reward), coinAmount = parseInteger(body.coinAmount); if (!Number.isSafeInteger(reward) || reward <= 0 || !Number.isSafeInteger(coinAmount) || coinAmount < reward) return fail("奖励存档数据无效"); // The client calculates and reports the award and resulting balance. // Save both the receipt and balance atomically; retries never rewrite a newer balance. stage.reward = reward; stage.rewardSaved = true; reportedBalance = coinAmount; } } else if (action === "start_stage") { if (requestedStage === run.stage && run.stages.some((s: any) => s.stage === requestedStage)) { // The same button request was already applied, including after its response was lost. } else { if (run.status !== "waiting" || requestedStage !== run.stage + 1 || requestedStage > 3) return fail("下一阶段尚未解锁"); const stage = await makeStage(requestedStage, matchingStartLevel(body, user), onlyId, now, run.pools[requestedStage - 1]); run.stages.push(stage); run.stage = requestedStage; run.status = "playing"; } } else { // An old result retried after the next stage began must never affect the new stage. const old = run.stages.find((s: any) => s.stage === requestedStage); if (!old) return fail("阶段不匹配"); const completed = old.results.some((r: any) => r.id === body.attemptId); if (!completed) { if (requestedStage !== run.stage) return fail("阶段已经结束"); if (action === "begin") Rules.beginAttempt(run, body.attemptId); else Rules.finishAttempt(run, body.attemptId, body.outcome, now); } } } // A slow match must not publish a playing state beyond the personal deadline. Rules.expire(state.run, Date.now()); const events = Analytics.collect(JSON.parse(before).run, state.run, body.isDebug ?? user.isDebug, now); if (events.length) state.analyticsQueue = [...(state.analyticsQueue || []), ...events]; const serialized = JSON.stringify(state); if (serialized !== before) { const update: any = { $set: { cloudRiseState: serialized } }; if (reportedBalance !== undefined) Object.assign(update.$set, { coinAmount: reportedBalance, timestamp: now }); const result = await users().updateOne(oldStateQuery(user), update); if (result.modifiedCount !== 1) continue; user.cloudRiseState = serialized; if (reportedBalance !== undefined) user.coinAmount = reportedBalance; } let sampleSyncPending = false; try { await persistHistory(uid, state.run); } catch (error) { sampleSyncPending = true; console.error("cloudRise history pending", error); } // Events and gameplay were committed atomically. A failed send must not fail settlement. try { const delivered = await Analytics.send(state.analyticsQueue || [], user); if (delivered.length) { const remaining = (state.analyticsQueue || []).filter((event: any) => !delivered.includes(event.id)); const acknowledged = { ...state }; if (remaining.length) acknowledged.analyticsQueue = remaining; else delete acknowledged.analyticsQueue; // Never overwrite progress committed by a concurrent request while analytics was in flight. await users().updateOne(oldStateQuery(user), { $set: { cloudRiseState: JSON.stringify(acknowledged) } }); } } catch (error) { console.error("cloudRise analytics pending", error instanceof Error ? error.message : error); } return response(user, state, config, Date.now(), sampleSyncPending); } return fail("活动状态正在更新,请重试"); } catch (error) { console.error("cloudRise", error); return fail(error instanceof Error ? error.message : "活动暂时不可用,请稍后重试"); } }