391 lines
22 KiB
TypeScript
391 lines
22 KiB
TypeScript
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<any> {
|
|
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<any> {
|
|
const profiles = await users().aggregate([
|
|
{ $match: { onlyId: { $ne: onlyId }, taskTime: { $gt: 0, $lt: inactiveBefore(now) } } },
|
|
{ $sample: { size: 500 } },
|
|
{ $project: { _id: 0, username: 1, useravatar: 1, useravatarIcon: 1 } },
|
|
]).toArray();
|
|
const pool = profiles.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<string, number[]> = {};
|
|
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<void> {
|
|
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<any> {
|
|
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<void> {
|
|
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 : "活动暂时不可用，请稍后重试");
|
|
}
|
|
}
|