diff --git a/server/laf-cloud/functions/rechargeStats.ts b/server/laf-cloud/functions/rechargeStats.ts index 7b46e3c..69c9a85 100644 --- a/server/laf-cloud/functions/rechargeStats.ts +++ b/server/laf-cloud/functions/rechargeStats.ts @@ -7,7 +7,7 @@ const BATCH_SIZE = 100; function emptyStats(asOf: number) { return { - version: 3, + version: 4, currency: 'CNY', unit: 'fen', asOf, @@ -44,8 +44,10 @@ function timestamp(value: any): number | null { export default async function () { const mongo = cloud.mongo.db; const users = mongo.collection('users'); + const rechargeStatsCollection = mongo.collection('userRechargeStats'); + const updatedAt = Date.now(); // End of yesterday in Beijing time, independent of execution time and server timezone. - const asOf = Math.floor((Date.now() + REPORT_TIMEZONE_OFFSET) / DAY) * DAY - REPORT_TIMEZONE_OFFSET - 1; + const asOf = Math.floor((updatedAt + REPORT_TIMEZONE_OFFSET) / DAY) * DAY - REPORT_TIMEZONE_OFFSET - 1; let afterId: any; let processedUsers = 0; let updatedUsers = 0; @@ -119,30 +121,30 @@ export default async function () { summary.stats.amount30d = summary.fen30d; summary.stats.amountTotal = summary.fenTotal; } - const result = await users.bulkWrite(batch.map(user => { + const result = await rechargeStatsCollection.bulkWrite(batch.map(user => { const rechargeStats = summaries.get(user.openid)?.stats ?? { ...emptyStats(asOf), missingOpenid: true }; + const record = { _id: user._id, openid: user.openid ?? null, ...rechargeStats, updatedAt }; return { updateOne: { - // Recheck eligibility and identity; an earlier reporting day cannot replace a later one. - filter: { - _id: user._id, - openid: user.openid ?? null, - pay_user: true, - $or: [ - // Earlier versions could have a later cutoff: allow their first full recalculation. - { 'rechargeStats.version': { $in: [1, 2] } }, - { 'rechargeStats.asOf': { $exists: false } }, - { 'rechargeStats.asOf': { $lte: asOf } }, - ], - }, - update: { $set: { rechargeStats } }, + // Match only the unique user ID, so a newer snapshot cannot cause an upsert ID conflict. + // Apply the freshness check atomically inside the update, including same-day reruns. + filter: { _id: user._id }, + update: [{ $replaceWith: { $cond: [ + { $and: [ + { $lte: [{ $ifNull: ['$asOf', 0] }, asOf] }, + { $lte: [{ $ifNull: ['$updatedAt', 0] }, updatedAt] }, + ] }, + { $literal: record }, + '$$ROOT', + ] } }], + upsert: true, } }; })); processedUsers += batch.length; - updatedUsers += result.modifiedCount; + updatedUsers += result.modifiedCount + result.upsertedCount; afterId = batch[batch.length - 1]._id; } - const result = { asOf, processedUsers, updatedUsers }; + const result = { asOf, updatedAt, processedUsers, updatedUsers }; console.log('rechargeStats completed', result); return { code: 1, data: result, msg: '充值统计完成' }; } diff --git a/server/laf-cloud/functions/rechargeStats.yaml b/server/laf-cloud/functions/rechargeStats.yaml index 83c95da..a121e02 100644 --- a/server/laf-cloud/functions/rechargeStats.yaml +++ b/server/laf-cloud/functions/rechargeStats.yaml @@ -1,5 +1,5 @@ name: rechargeStats -desc: "付费用户充值统计（分）；每日03:00执行，固定截至北京时间昨日23:59:59.999" +desc: "付费用户充值统计写入userRechargeStats（分）；每日03:00执行，固定截至北京时间昨日23:59:59.999" methods: [] tags: - statistics diff --git a/server/laf-cloud/rechargeStats.README.md b/server/laf-cloud/rechargeStats.README.md index 0d0db81..02c3a0c 100644 --- a/server/laf-cloud/rechargeStats.README.md +++ b/server/laf-cloud/rechargeStats.README.md @@ -1,26 +1,37 @@ # 付费用户充值统计定时任务 -云函数 `rechargeStats` 每次全量重算当前 `users.pay_user === true` 的玩家，将结果覆盖到 `users.rechargeStats`。每日凌晨 03:00 执行，统计截止时间固定为北京时间前一天 `23:59:59.999`；不修改订单和玩家付费标记。 +云函数 `rechargeStats` 每次全量重算读取时 `users.pay_user === true` 的玩家，将结果写入独立集合 `userRechargeStats`。每位玩家只保留一条最新结果，使用与 `users._id` 相同的值和 BSON 类型作为 `_id`，重复执行时覆盖，不追加每日历史。每日凌晨 03:00 执行，统计截止时间固定为北京时间前一天 `23:59:59.999`；`users` 和 `order` 均只读。 ## 保存字段 | 字段 | 含义 | | --- | --- | +| `_id` | 对应 `users._id`，保留原始类型，作为唯一关联键 | +| `openid` | 本轮读取的玩家 openid；缺失时为 null | | `amount15d` | 从统计截止时间倒推 15 × 24 小时的充值金额，整数分 | | `amount30d` | 从统计截止时间倒推 30 × 24 小时的充值金额，整数分 | | `amountTotal` | 截至统计截止时间（含）、当前库中可统计的累计充值金额，整数分 | | `currency` | 固定 `CNY` | | `unit` | 固定 `fen`，金额单位为分 | | `asOf` | 本轮执行日期的北京时间前一天 `23:59:59.999`，原始 Unix 毫秒时间戳 | +| `updatedAt` | 本轮开始计算的 Unix 毫秒时间戳，所有批次相同，并非每条写入完成时间 | | `orderCount` | 纳入累计金额的去重订单数 | | `fallbackTimeOrderCount` | 无支付确认时间，使用下单时间的订单数 | | `missingTimeOrderCount` | 支付和下单时间均缺失，仅计入累计的订单数 | | `invalidOrderCount` | 缺失订单号、金额或数量无效而跳过的记录数 | | `duplicateOrderCount` | 跳过的重复订单记录数 | | `missingOpenid` | 用户是否缺少可关联订单的 openid | -| `version` | 统计规则版本，当前为 3 | +| `version` | 统计规则及存储版本，当前为 4 | -版本 1 的金额单位为元，版本 2 使用分单位但按运行时间减 3 小时作为截止点。版本 3 首次运行会重新计算并整体覆盖版本 1/2 的统计，即使旧版 `asOf` 晚于新版截止时间，也允许完成升级。读取方应以 `unit: "fen"` 识别分单位，以 `version: 3` 识别固定前一天末尾的截止口径；运行未覆盖到的用户仍可能保留旧版数据，不能仅按字段名判断单位。 +版本 1 的金额单位为元，版本 2 使用分单位但按运行时间减 3 小时作为截止点，版本 3 改为固定前一天末尾。版本 4 保留版本 3 的计算口径，改为独立集合。首次运行从订单重算，不读取或复制 `users.rechargeStats`，旧字段的单位和截止点不影响新结果。 + +金额字段直接位于新文档根层级，查询示例： + +```javascript +// userId 必须与 users._id 类型一致；ObjectId 不要转成字符串。 +const stats = await db.collection('userRechargeStats').findOne({ _id: userId }); +// stats.amount15d / stats.amount30d / stats.amountTotal，单位均为分。 +``` ## 统计口径 @@ -37,9 +48,13 @@ ## 执行与一致性 -用户按 `_id` 游标每批 100 条读取，订单使用 MongoDB 游标遍历，避免默认查询上限截断。每批只更新 `rechargeStats`，写入时重新检查 `pay_user` 和 `openid`，且较早统计日期不会覆盖较新 `asOf` 的快照。同一天重跑会覆盖该天统计，不重复累加；跨到下一个北京时间日期时，旧订单会按新截止时间退出滚动窗口。 +用户按 `_id` 游标每批 100 条读取，只投影 `_id`、`openid`，订单使用 MongoDB 游标遍历，避免默认查询上限截断。每批向 `userRechargeStats` 批量 upsert：不存在则新增，存在则整体替换统计文档。文档用于统计任务专有数据，不应混入其他业务字段。不会逐个用户查询或写回 `users`，但每轮仍需读取付费名单及其历史订单，独立集合不消除这些计算开销。 -整轮不是数据库事务：运行期间新支付或用户变更可能下一轮才体现。失败时抛出错误，已完成批次保留，剩余用户下次重算；每个用户的 `asOf` 可判断新旧结果，不能把混合轮次当作同一时刻的全库快照。 +写入通过 MongoDB 更新管道原子比较 `asOf` 和 `updatedAt`：较早统计日期或较早启动的任务不能覆盖较新的结果，包括同一天并发执行的情况。同一天重跑会重新计算，不重复累加；跨到下一个北京时间日期时，旧订单会按新截止时间退出滚动窗口。需 MongoDB 4.2 或以上支持更新管道，参考 [MongoDB 官方文档](https://www.mongodb.com/docs/manual/tutorial/update-documents-with-aggregation-pipeline/)。 + +付费标记和 openid 以每批读取时为准，写入独立集合时不再重新检查 `users`。整轮不是数据库事务：运行期间新支付或用户变更可能下一轮才体现。失败时抛出错误，已完成批次保留，剩余用户下次重算。用户被删除或付费标记改为 false 后，不会自动删除其已有统计，新任务会跳过该用户，保留旧的 `asOf` / `updatedAt`。 + +返回值中 `processedUsers` 是本轮已处理人数，`updatedUsers` 包括新增及实际修改的记录，跳过较新结果或完全相同的记录不计入。成功返回的 `updatedAt` 可用于筛选本轮写入的结果；任务失败时不能将已完成批次视为完整快照。`asOf` 相同也可能来自不同轮次，需结合 `updatedAt` 判断。 建议上线前在 Laf 数据库控制台建立以下非唯一索引（若已有等价索引则复用）： @@ -48,12 +63,14 @@ db.collection('users').createIndex({ pay_user: 1, _id: 1 }); db.collection('order').createIndex({ openid: 1, outTradeNo: 1, _id: 1 }); ``` +`userRechargeStats` 自带的唯一 `_id` 索引满足按用户写入和查询。按金额排序、按执行轮次筛选等索引应根据实际查询增加。本地代码不会自动创建上述生产索引；两个集合仍共享所在数据库实例的 CPU、内存和磁盘资源。 + ## 发布和启用 代码与触发器配置是独立资源，提交本地文件不会自动启动线上定时任务。`recharge-stats.trigger.json` 使用 Laf 创建触发器 API 的 `desc / target / cron` 字段。 -1. 在目标 Laf 应用发布 `functions/rechargeStats.ts` 和同名 YAML，保持 `methods: []`，不开放公共 HTTP 入口。 -2. 在云函数控制台手动执行一次 `rechargeStats`（无参数），检查返回的 `processedUsers`、`updatedUsers` 和部分玩家的结果，确认运行耗时。 +1. 停止旧版任务并等待正在运行的旧任务结束，在目标 Laf 应用发布 `functions/rechargeStats.ts` 和同名 YAML，保持 `methods: []`，不开放公共 HTTP 入口。 +2. 在云函数控制台手动执行一次 `rechargeStats`（无参数），首次 upsert 会创建 `userRechargeStats` 集合。检查返回的 `processedUsers`、`updatedUsers`，抽查新集合中的金额、`asOf`、`updatedAt`、`version: 4` 和用户关联，确认运行耗时。验证后将后台或其他读取方切换到新集合；旧 `users.rechargeStats` 不再更新，但本任务不自动删除该字段，清理可在确认所有读取方完成切换后另行执行。 3. 在触发器面板绑定函数 `rechargeStats`，使用 `recharge-stats.trigger.json` 的配置：每日 03:00 执行。表达式 `0 3 * * *` 按触发器时区解释；目标是北京时间 03:00，应确认调度时区为 `Asia/Shanghai`。如果部署使用 UTC 调度，则使用 `0 19 * * *`（UTC 19:00 为次日北京时间 03:00）。统计日期固定按北京时间计算，与订单存储的 8 小时修正分别处理。 4. 检查已有触发器，避免重复创建。若使用已登录且已绑定正确应用的 Laf CLI，可执行： @@ -62,7 +79,7 @@ laf trigger list laf trigger create "付费用户充值统计（每日凌晨3点）" rechargeStats "0 3 * * *" ``` -5. 首次定时触发后确认日志出现 `rechargeStats completed`，抽查 `users.rechargeStats.asOf` 已更新。大规模历史数据上线前应确认全量耗时在云函数执行时限内。 +5. 首次定时触发后确认日志出现 `rechargeStats completed`，抽查 `userRechargeStats` 中 `asOf` 和 `updatedAt` 已更新。大规模历史数据上线前应确认全量耗时在云函数执行时限内。 参考：[Laf 定时任务文档](https://doc.laf.run/zh/cloud-function/cron.html)、[官方 CLI 触发器命令](https://github.com/labring/laf/blob/main/cli/src/command/trigger/index.ts)、[创建触发器字段](https://github.com/labring/laf/blob/main/server/src/trigger/dto/create-trigger.dto.ts)。 @@ -74,4 +91,4 @@ Node.js 24.11 或兼容 `registerHooks` / `stripTypeScriptTypes` 的版本： node --test laf-cloud/tests/recharge-stats.test.mjs ``` -覆盖分单位、固定北京时间前一天末尾截止、15/30 天毫秒边界、准时/延迟/手动执行、跨月/跨年/闰日、当天订单次日计入、版本 1/2 统计升级、8 小时存储修正、支付时间优先、时间回退、无效金额、数量、去重、测试订单排除、超过 1000 条订单、多页用户、重跑、跨统计日期写保护、异常中断和关闭 HTTP 入口。使用内存 MongoDB 接口替身，不连接生产数据库。 +覆盖分单位、固定北京时间前一天末尾截止、15/30 天毫秒边界、准时/延迟/手动执行、跨月/跨年/闰日、当天订单次日计入、旧嵌入字段不影响新结果、8 小时存储修正、支付时间优先、时间回退、无效金额、数量、去重、测试订单排除、超过 1000 条订单、多页用户、独立集合 upsert、users 只读、重跑、跨日期及同日并发写保护、异常中断和关闭 HTTP 入口。使用内存 MongoDB 接口替身，不连接生产数据库；上线仍需实际 MongoDB 环境验证。 diff --git a/server/laf-cloud/tests/recharge-stats.test.mjs b/server/laf-cloud/tests/recharge-stats.test.mjs index 991daa7..4f7ce7f 100644 --- a/server/laf-cloud/tests/recharge-stats.test.mjs +++ b/server/laf-cloud/tests/recharge-stats.test.mjs @@ -7,7 +7,7 @@ const DAY = 86400000; const OFFSET = 8 * 3600000; const NOW = Date.parse('2026-09-15T03:00:00+08:00'); const CUTOFF = Date.parse('2026-09-14T23:59:59.999+08:00'); -const state = { users: [], order: [], writes: 0, closed: 0, beforeWrite: null, failRead: false }; +const state = { users: [], order: [], userRechargeStats: [], reads: [], writes: 0, closed: 0, beforeWrite: null, failRead: false }; const realNow = Date.now; Date.now = () => NOW; test.after(() => { Date.now = realNow; }); @@ -30,11 +30,26 @@ function matches(row, query) { }); } +function evaluate(expression, row) { + if (expression === '$$ROOT') return row; + if (typeof expression === 'string' && expression.startsWith('$')) return row[expression.slice(1)]; + if (!expression || typeof expression !== 'object') return expression; + if ('$literal' in expression) return expression.$literal; + const [operator, values] = Object.entries(expression)[0]; + if (operator === '$cond') return evaluate(values[evaluate(values[0], row) ? 1 : 2], row); + if (operator === '$and') return values.every(value => evaluate(value, row)); + if (operator === '$ifNull') return evaluate(values[0], row) ?? evaluate(values[1], row); + if (operator === '$lte') return evaluate(values[0], row) <= evaluate(values[1], row); + throw new Error(`Unsupported update expression: ${operator}`); +} + globalThis.__rechargeStatsCloud = { mongo: { db: { collection(name) { - assert.ok(['users', 'order'].includes(name), 'must not count iosOrder temporary records'); + assert.ok(['users', 'order', 'userRechargeStats'].includes(name), 'must not count iosOrder temporary records'); return { - find(query) { + find(query, options) { + state.reads.push(name); + if (name === 'users') assert.deepEqual(options.projection, { _id: 1, openid: 1 }); let rows = state[name].filter(row => matches(row, query)); return { sort(sort) { @@ -48,7 +63,9 @@ globalThis.__rechargeStatsCloud = { mongo: { db: { return this; }, limit(size) { rows = rows.slice(0, size); return this; }, - async toArray() { return rows.map(row => ({ ...row })); }, + async toArray() { return rows.map(row => Object.fromEntries( + Object.keys(options.projection).filter(key => row[key] !== undefined).map(key => [key, row[key]]) + )); }, async *[Symbol.asyncIterator]() { if (state.failRead) throw new Error('database read failed'); yield* rows; @@ -57,17 +74,24 @@ globalThis.__rechargeStatsCloud = { mongo: { db: { }; }, async bulkWrite(operations) { + assert.equal(name, 'userRechargeStats', 'users and order must remain read-only'); state.writes++; state.beforeWrite?.(); let modifiedCount = 0; + let upsertedCount = 0; for (const { updateOne } of operations) { - const user = state.users.find(row => matches(row, updateOne.filter)); - if (user) { - Object.assign(user, structuredClone(updateOne.update.$set)); - modifiedCount++; + assert.deepEqual(Object.keys(updateOne.filter), ['_id']); + assert.equal(updateOne.upsert, true); + assert.equal(updateOne.update.length, 1); + const index = state.userRechargeStats.findIndex(row => matches(row, updateOne.filter)); + const previous = index < 0 ? { _id: updateOne.filter._id } : state.userRechargeStats[index]; + const next = structuredClone(evaluate(updateOne.update[0].$replaceWith, previous)); + if (index < 0) { state.userRechargeStats.push(next); upsertedCount++; } + else if (JSON.stringify(previous) !== JSON.stringify(next)) { + state.userRechargeStats[index] = next; modifiedCount++; } } - return { modifiedCount }; + return { modifiedCount, upsertedCount }; }, }; }, @@ -87,10 +111,11 @@ const { default: run } = await import('../functions/rechargeStats.ts'); function reset() { Object.assign(state, { - users: [{ _id: 'u1', openid: 'o1', pay_user: true }], order: [], + users: [{ _id: 'u1', openid: 'o1', pay_user: true }], order: [], userRechargeStats: [], reads: [], writes: 0, closed: 0, beforeWrite: null, failRead: false, }); } +function statsFor(id = 'u1') { return state.userRechargeStats.find(row => row._id === id); } function order(id, age, extra = {}) { return { _id: id, outTradeNo: id, openid: 'o1', state: 2, goodsPrice: 100, itemCount: 1, @@ -106,14 +131,18 @@ test('windows exclude the previous end-of-day boundary and include yesterday fin order('early-morning', -2 * 3600000), order('future', -4 * 3600000), ]; await run(); - const stats = state.users[0].rechargeStats; + const stats = statsFor(); assert.equal(stats.amount15d, 200); assert.equal(stats.amount30d, 400); assert.equal(stats.amountTotal, 500); assert.equal(stats.orderCount, 5); assert.equal(stats.asOf, CUTOFF); - assert.equal(stats.version, 3); + assert.equal(stats.version, 4); assert.equal(stats.unit, 'fen'); + assert.equal(stats.updatedAt, NOW); + assert.equal(stats._id, 'u1'); + assert.equal(stats.openid, 'o1'); + assert.equal(state.users[0].rechargeStats, undefined); }); test('counts quantities, numeric strings, paid states and deduplicates; excludes unpaid and test orders', async () => { @@ -126,9 +155,9 @@ test('counts quantities, numeric strings, paid states and deduplicates; excludes order('test-env', DAY, { paymentAppEnv: 'test' }), order('wct_test', DAY), ]; await run(); - assert.equal(state.users[0].rechargeStats.amountTotal, 310); - assert.equal(state.users[0].rechargeStats.orderCount, 2); - assert.equal(state.users[0].rechargeStats.duplicateOrderCount, 1); + assert.equal(statsFor().amountTotal, 310); + assert.equal(statsFor().orderCount, 2); + assert.equal(statsFor().duplicateOrderCount, 1); }); test('falls back to unshifted creation time, tracks unknown times, and rejects malformed amounts', async () => { @@ -142,7 +171,7 @@ test('falls back to unshifted creation time, tracks unknown times, and rejects m order('overflow', 0, { goodsPrice: Number.MAX_SAFE_INTEGER, itemCount: 2 }), ]; await run(); - const stats = state.users[0].rechargeStats; + const stats = statsFor(); assert.equal(stats.amount15d, 100); assert.equal(stats.amount30d, 200); assert.equal(stats.amountTotal, 300); @@ -161,60 +190,94 @@ test('processes every page, restricts to boolean true, resets old totals, and fl state.users.push({ _id: 'free', openid: 'free', pay_user: false }); state.users.push({ _id: 'string', openid: 'string', pay_user: 'true' }); state.order = [order('last', 1, { openid: 'o250' }), order('free', 1, { openid: 'free' })]; + state.userRechargeStats = [{ _id: 'u0000', asOf: CUTOFF - DAY, amountTotal: 999 }]; + const originalUsers = structuredClone(state.users); const result = await run(); assert.equal(result.data.processedUsers, 252); + assert.equal(result.data.updatedUsers, 252); assert.equal(state.writes, 3); - assert.equal(state.users.find(user => user.openid === 'o250').rechargeStats.amountTotal, 100); - assert.equal(state.users.find(user => user.openid === 'o0').rechargeStats.amountTotal, 0); - assert.equal(state.users.find(user => user._id === 'no-openid').rechargeStats.missingOpenid, true); - assert.equal(state.users.find(user => user._id === 'free').rechargeStats, undefined); - assert.equal(state.users.find(user => user._id === 'string').rechargeStats, undefined); + assert.equal(statsFor('u0250').amountTotal, 100); + assert.equal(statsFor('u0000').amountTotal, 0); + assert.equal(statsFor('no-openid').missingOpenid, true); + assert.equal(statsFor('free'), undefined); + assert.equal(statsFor('string'), undefined); + assert.equal(state.userRechargeStats.length, 252); + assert.equal(state.reads.filter(name => name === 'users').length, 4, 'one read per page, not per user'); + assert.deepEqual(state.users, originalUsers); }); test('reruns overwrite instead of accumulating; older runs cannot overwrite newer snapshots', async () => { reset(); state.order = [order('paid', 1)]; - await run(); - await run(); - assert.equal(state.users[0].rechargeStats.amountTotal, 100); - state.users[0].rechargeStats = { version: 3, asOf: CUTOFF + DAY, amountTotal: 200 }; - await run(); - assert.equal(state.users[0].rechargeStats.amountTotal, 200); + assert.equal((await run()).data.updatedUsers, 1, 'initial upsert counts as an update'); + assert.equal((await run()).data.updatedUsers, 0, 'identical rerun changes nothing'); + assert.equal(statsFor().amountTotal, 100); + assert.equal(state.userRechargeStats.length, 1); + state.userRechargeStats[0] = { _id: 'u1', version: 4, asOf: CUTOFF + DAY, updatedAt: NOW + DAY, amountTotal: 200 }; + assert.equal((await run()).data.updatedUsers, 0); + assert.equal(statsFor().amountTotal, 200); + assert.equal(state.userRechargeStats.length, 1); }); test('streams more than 1000 orders and expired windows clear on the next run', async () => { reset(); state.order = Array.from({ length: 1005 }, (_, i) => order(`paid-${i}`, 30 * DAY - 1, { goodsPrice: 1 })); await run(); - assert.equal(state.users[0].rechargeStats.amountTotal, 1005); - assert.equal(state.users[0].rechargeStats.amount30d, 1005); + assert.equal(statsFor().amountTotal, 1005); + assert.equal(statsFor().amount30d, 1005); try { Date.now = () => NOW + DAY; await run(); - assert.equal(state.users[0].rechargeStats.amount30d, 0); - assert.equal(state.users[0].rechargeStats.amountTotal, 1005); + assert.equal(statsFor().amount30d, 0); + assert.equal(statsFor().amountTotal, 1005); } finally { Date.now = () => NOW; } }); -test('rechecks user eligibility and openid when writing', async () => { +test('a newer same-day snapshot written during calculation is not overwritten or duplicated', async () => { reset(); - state.beforeWrite = () => { state.users[0].pay_user = false; }; - await run(); - assert.equal(state.users[0].rechargeStats, undefined); + state.order = [order('paid', DAY)]; + state.beforeWrite = () => { state.userRechargeStats = [ + { _id: 'u1', asOf: CUTOFF, updatedAt: NOW + 60000, amountTotal: 200 }, + ]; }; + assert.equal((await run()).data.updatedUsers, 0); + assert.equal(statsFor().amountTotal, 200); + assert.equal(statsFor().updatedAt, NOW + 60000); + assert.equal(state.userRechargeStats.length, 1); +}); + +test('an empty eligible set performs no writes and retains previous snapshots', async () => { reset(); - state.beforeWrite = () => { state.users[0].openid = 'changed'; }; - await run(); - assert.equal(state.users[0].rechargeStats, undefined); + state.users[0].pay_user = false; + state.userRechargeStats = [{ _id: 'u1', asOf: CUTOFF - DAY, updatedAt: NOW - DAY, amountTotal: 900 }]; + const previous = structuredClone(state.userRechargeStats); + const result = await run(); + assert.equal(result.data.processedUsers, 0); + assert.equal(result.data.updatedUsers, 0); + assert.equal(state.writes, 0); + assert.deepEqual(state.reads, ['users']); + assert.deepEqual(state.userRechargeStats, previous); +}); + +test('replacement preserves literal identity values and updates an existing same-day snapshot', async () => { + reset(); + state.users[0].openid = '$openid'; + state.order = [order('paid', DAY, { openid: '$openid' })]; + state.userRechargeStats = [{ _id: 'u1', asOf: CUTOFF, updatedAt: NOW - 60000, amountTotal: 900 }]; + assert.equal((await run()).data.updatedUsers, 1); + assert.equal(statsFor().openid, '$openid'); + assert.equal(statsFor().amountTotal, 100); + assert.equal(statsFor().updatedAt, NOW); + assert.equal(state.userRechargeStats.length, 1); }); test('read failure propagates, closes cursor and never replaces existing totals with zero', async () => { reset(); - state.users[0].rechargeStats = { asOf: NOW - 1, amountTotal: 9 }; + state.userRechargeStats = [{ _id: 'u1', asOf: CUTOFF - DAY, updatedAt: NOW - DAY, amountTotal: 9 }]; state.failRead = true; await assert.rejects(run(), /database read failed/); assert.equal(state.closed, 1); assert.equal(state.writes, 0); - assert.equal(state.users[0].rechargeStats.amountTotal, 9); + assert.equal(statsFor().amountTotal, 9); }); test('timer function does not expose a public HTTP endpoint', () => { @@ -225,19 +288,21 @@ test('timer function does not expose a public HTTP endpoint', () => { assert.equal(trigger.cron, '0 3 * * *'); }); -test('recalculates version 1 and 2 snapshots despite their later cutoffs', async () => { - for (const version of [1, 2]) { +test('ignores legacy embedded amounts and leaves users untouched while populating the new collection', async () => { + for (const version of [1, 2, 3]) { reset(); state.users[0].rechargeStats = { version, asOf: NOW - 1, amount15d: 1, amount30d: 1, amountTotal: 1 }; + const originalUser = structuredClone(state.users[0]); state.order = [order('paid', DAY, { goodsPrice: 101, itemCount: 3 })]; await run(); - const stats = state.users[0].rechargeStats; - assert.equal(stats.version, 3); + const stats = statsFor(); + assert.equal(stats.version, 4); assert.equal(stats.unit, 'fen'); assert.equal(stats.asOf, CUTOFF); assert.equal(stats.amount15d, 303); assert.equal(stats.amount30d, 303); assert.equal(stats.amountTotal, 303); + assert.deepEqual(state.users[0], originalUser); } }); @@ -245,13 +310,13 @@ test('payments between midnight and 03:00 are included on the following daily ru reset(); state.order = [order('midnight', -1), order('02:00', -2 * 3600000 - 1)]; await run(); - assert.equal(state.users[0].rechargeStats.amountTotal, 0); + assert.equal(statsFor().amountTotal, 0); try { Date.now = () => NOW + DAY; await run(); - assert.equal(state.users[0].rechargeStats.amount15d, 200); - assert.equal(state.users[0].rechargeStats.amount30d, 200); - assert.equal(state.users[0].rechargeStats.amountTotal, 200); + assert.equal(statsFor().amount15d, 200); + assert.equal(statsFor().amount30d, 200); + assert.equal(statsFor().amountTotal, 200); } finally { Date.now = () => NOW; } }); @@ -262,7 +327,7 @@ test('midnight, scheduled, delayed and manual runs on the same Beijing date have for (const time of ['00:00:00.000', '01:00:00.000', '03:00:00.000', '03:27:16.321', '12:00:00.000', '23:59:59.999']) { Date.now = () => Date.parse(`2026-09-15T${time}+08:00`); await run(); - const stats = state.users[0].rechargeStats; + const stats = statsFor(); assert.equal(stats.asOf, CUTOFF, time); assert.equal(stats.amount15d, 100, time); assert.equal(stats.amount30d, 100, time); @@ -281,7 +346,7 @@ test('Beijing midnight advances the cutoff across month, year and leap-day bound reset(); Date.now = () => Date.parse(runAt); await run(); - assert.equal(state.users[0].rechargeStats.asOf, Date.parse(cutoff)); + assert.equal(statsFor().asOf, Date.parse(cutoff)); } } finally { Date.now = () => NOW; } });