import cloud from '@lafjs/cloud'; import { buildPaymentProfile, paymentVipLevel, PAYMENT_PROFILE_RULE } from '@/paymentProfile'; import type { ProfileOrder } from '@/paymentProfile'; const DAY = 24 * 60 * 60 * 1000; const LEGACY_TIME_OFFSET = 8 * 60 * 60 * 1000; const REPORT_TIMEZONE_OFFSET = 8 * 60 * 60 * 1000; // Asia/Shanghai const BATCH_SIZE = 100; const userKey = (id: any) => `${typeof id}:${String(id)}`; function emptyStats(asOf: number) { return { version: 8, currency: 'CNY', unit: 'fen', asOf, amount15d: 0, amount30d: 0, amountTotal: 0, amountMax: 0, orderCount: 0, fallbackTimeOrderCount: 0, missingTimeOrderCount: 0, invalidOrderCount: 0, duplicateOrderCount: 0, duplicateConflictCount: 0, missingOpenid: false, }; } function positiveInteger(value: any): number | null { if (typeof value !== 'number' && typeof value !== 'string') return null; const number = Number(value); return Number.isSafeInteger(number) && number > 0 ? number : null; } function timestamp(value: any): number | null { if (value instanceof Date) return positiveInteger(value.getTime()); const numeric = positiveInteger(value); if (numeric !== null) return numeric; // Date strings must include a timezone; do not depend on the server's timezone. if (typeof value === 'string' && /T.*(?:Z|[+-]\d{2}:?\d{2})$/i.test(value)) { return positiveInteger(Date.parse(value)); } return null; } // Laf timer entry point. HTTP methods are disabled in rechargeStats.yaml. 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((updatedAt + REPORT_TIMEZONE_OFFSET) / DAY) * DAY - REPORT_TIMEZONE_OFFSET - 1; let afterId: any; let processedUsers = 0; let updatedUsers = 0; let updatedVipUsers = 0; let validProfiles = 0; let invalidProfiles = 0; let scannedUsers = 0; let skippedNeverPaidUsers = 0; while (true) { const candidates: any[] = await users.find({ ...(afterId === undefined ? {} : { _id: { $gt: afterId } }), }, { projection: { _id: 1, openid: 1, pay_user: 1, register_time: 1, vip_level: 1, vip_profile: 1 } }).sort({ _id: 1 }).limit(BATCH_SIZE).toArray(); if (!candidates.length) break; // Advance on the identity page, including pages consisting entirely of stable VIP0 users. afterId = candidates[candidates.length - 1]._id; scannedUsers += candidates.length; const previous: any[] = await rechargeStatsCollection.find({ _id: { $in: candidates.map(user => user._id) } }, { projection: { _id: 1, openid: 1, has_ever_paid: 1, paid_status: 1, data_valid: 1, version: 1, asOf: 1, updatedAt: 1, profile_rule_version: 1, maintenance_identity: 1 } }).toArray(); const previousById = new Map(previous.map(row => [userKey(row._id), row])); const batch = candidates.filter(user => { const saved = previousById.get(userKey(user._id)); // Only a verified, initialized never-payer can be cached. A missing profile, // ambiguous identity or changed account must go through the order check below. const stableNeverPaid = user.pay_user === false && saved?.version === 8 && saved.data_valid === true && saved.paid_status === 'never' && saved.has_ever_paid === false && user.vip_level === 0 && user.vip_profile?.data_valid === true && user.vip_profile.as_of === saved.asOf && user.vip_profile.updated_at === saved.updatedAt && user.vip_profile.rule_version === PAYMENT_PROFILE_RULE.version && saved.profile_rule_version === PAYMENT_PROFILE_RULE.version && saved.openid === user.openid && saved.maintenance_identity?.pay_user === false && saved.maintenance_identity.register_time === timestamp(user.register_time); if (stableNeverPaid) skippedNeverPaidUsers++; return !stableNeverPaid; }); if (!batch.length) continue; const previouslyPaid = new Set([ ...previous.filter(row => row.has_ever_paid === true), ...batch.filter(user => user.vip_level > 0 || user.vip_profile?.has_ever_paid === true), ].map(row => userKey(row._id))); const summaries = new Map; seen: Map; fen15d: number; fen30d: number; fenTotal: number; recent: ProfileOrder[]; lastPaidAt: number | null; futureOrders: number; }>(); for (const user of batch) { if (typeof user.openid === 'string' && user.openid.trim() && !summaries.has(user.openid)) { summaries.set(user.openid, { stats: emptyStats(asOf), seen: new Map(), fen15d: 0, fen30d: 0, fenTotal: 0, recent: [], lastPaidAt: null, futureOrders: 0 }); } } if (summaries.size) { const orders = mongo.collection('order').find({ openid: { $in: [...summaries.keys()] }, state: { $in: [1, 2] }, paymentAppEnv: { $ne: 'test' }, outTradeNo: { $not: /^wct_/ }, }, { projection: { openid: 1, outTradeNo: 1, goodsPrice: 1, itemCount: 1, chargeTime: 1, time: 1, itemid: 1, 'goldMiner.configVersion': 1, 'goldMiner.paidAt': 1, 'goldMiner.confirmedAt': 1, rewardVersion: 1, activityType: 1 } }) .sort({ openid: 1, outTradeNo: 1, _id: 1 }); try { for await (const order of orders) { const summary = summaries.get(order.openid)!; const price = positiveInteger(order.goodsPrice); const count = positiveInteger(order.itemCount); const amount = price === null || count === null ? NaN : price * count; if (typeof order.outTradeNo !== 'string' || !order.outTradeNo.trim() || !Number.isSafeInteger(amount)) { summary.stats.invalidOrderCount++; continue; } // Existing payment writers store new Date(Date.now() + 8h). // order.time is the original, unshifted Unix timestamp. const chargeTime = timestamp(order.chargeTime); const confirmedAt = timestamp(order.goldMiner?.paidAt) ?? timestamp(order.goldMiner?.confirmedAt) ?? (chargeTime !== null && chargeTime > LEGACY_TIME_OFFSET ? chargeTime - LEGACY_TIME_OFFSET : null); const paidAt = confirmedAt ?? timestamp(order.time); if (paidAt !== null && paidAt > asOf) { summary.futureOrders++; continue; } if (summary.seen.has(order.outTradeNo)) { summary.stats.duplicateOrderCount++; const first = summary.seen.get(order.outTradeNo)!; if (first.amount !== amount || first.paidAt !== paidAt) summary.stats.duplicateConflictCount++; continue; } summary.seen.set(order.outTradeNo, { amount, paidAt }); summary.fenTotal += amount; if (!Number.isSafeInteger(summary.fenTotal)) throw new Error('Recharge total exceeds safe integer range'); summary.stats.orderCount++; summary.stats.amountMax = Math.max(summary.stats.amountMax, amount); if (paidAt === null) summary.stats.missingTimeOrderCount++; else { summary.lastPaidAt = Math.max(summary.lastPaidAt ?? 0, paidAt); if (confirmedAt === null) summary.stats.fallbackTimeOrderCount++; if (paidAt > asOf - 15 * DAY) summary.fen15d += amount; if (paidAt > asOf - 30 * DAY) { summary.fen30d += amount; summary.recent.push({ amountFen: amount, paidAt, sku: String(order.itemid ?? ''), rewardVersion: order.rewardVersion ?? order.goldMiner?.configVersion, activity: order.activityType ?? (order.goldMiner ? 'coin' : undefined) }); } } } } finally { await orders.close(); } } for (const summary of summaries.values()) { summary.stats.amount15d = summary.fen15d; summary.stats.amount30d = summary.fen30d; summary.stats.amountTotal = summary.fenTotal; } const result = await rechargeStatsCollection.bulkWrite(batch.map(user => { const rechargeStats = summaries.get(user.openid)?.stats ?? { ...emptyStats(asOf), missingOpenid: true }; const summary = summaries.get(user.openid); const profile = buildPaymentProfile({ cutoff: asOf + 1, registerTime: timestamp(user.register_time), payUser: user.pay_user, missingOpenid: rechargeStats.missingOpenid, previousPaid: previouslyPaid.has(userKey(user._id)), confirmedOrders: rechargeStats.orderCount, futureOrders: summary?.futureOrders ?? 0, invalidOrders: rechargeStats.invalidOrderCount, missingTimeOrders: rechargeStats.missingTimeOrderCount, fallbackTimeOrders: rechargeStats.fallbackTimeOrderCount, duplicateConflicts: rechargeStats.duplicateConflictCount, amount15d: rechargeStats.amount15d, amount30d: rechargeStats.amount30d, lastPaidAt: summary?.lastPaidAt ?? null, recentOrders: summary?.recent ?? [], }); if (profile.data_valid) validProfiles++; else invalidProfiles++; // Persist calculation inputs/results here; the public VIP attribute belongs only to users. const { vip_level: _vipLevel, ...features } = profile; const record = { _id: user._id, openid: user.openid ?? null, ...rechargeStats, ...features, updatedAt, maintenance_identity: { pay_user: user.pay_user ?? null, register_time: timestamp(user.register_time) } }; // Preserve the complete last calculated feature tuple atomically on bad input. // Keep current error/identity metadata and its attempted rule version visible. const heldFields = Object.fromEntries([ 'profile_rule_version', 'profile_calculated_as_of', 'rfm', 'spending', 'ticket_factors', ].map(key => [key, { $ifNull: [`$${key}`, null] }])); let replacement: any = profile.data_valid ? { $literal: record } : { $mergeObjects: [{ $literal: record }, heldFields] }; // A concurrent earlier run may establish paid identity after our batch read. // Do not erase it or publish VIP0 over it, even on structurally valid free-user input. if (profile.has_ever_paid !== true) replacement = { $cond: [ { $eq: ['$has_ever_paid', true] }, { $mergeObjects: [{ $literal: { ...record, has_ever_paid: true, paid_status: 'unknown', data_valid: false, invalid_reasons: [...profile.invalid_reasons, 'paid_identity_changed_during_calculation'] } }, heldFields] }, replacement, ] }; return { updateOne: { // 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] }, ] }, replacement, '$$ROOT', ] } }], upsert: true, } }; })); // Read back the committed tuple: a concurrent task may have won the freshness check, // or invalid input may have retained a newer good tuple. Never publish the attempted result. const committed: any[] = await rechargeStatsCollection.find({ _id: { $in: batch.map(user => user._id) } }, { projection: { _id: 1, openid: 1, asOf: 1, updatedAt: 1, data_valid: 1, paid_status: 1, has_ever_paid: 1, maintenance_identity: 1, profile_rule_version: 1, profile_calculated_as_of: 1, rfm: 1 } }).toArray(); const batchById = new Map(batch.map(user => [userKey(user._id), user])); const vipResult = committed.length ? await users.bulkWrite(committed.map(saved => { const source = batchById.get(userKey(saved._id))!; // The retained RFM tuple also permits first-time migration/retry without the legacy // userRechargeStats.vip_level field, even if this attempt has invalid order inputs. const score = saved.rfm?.score; const calculatedLevel = Number.isFinite(score) && saved.profile_calculated_as_of != null && saved.rfm.last_effective_paid_at !== undefined ? paymentVipLevel(saved.rfm.last_effective_paid_at !== null, score) : null; const identity = { openid: saved.openid ?? null, pay_user: saved.maintenance_identity?.pay_user ?? null, register_time: saved.maintenance_identity?.register_time ?? null }; const currentIdentity = { $and: [ { $eq: [{ $ifNull: ['$openid', null] }, { $literal: identity.openid ?? null }] }, { $eq: [{ $ifNull: ['$pay_user', null] }, { $literal: identity.pay_user ?? null }] }, { $eq: [{ $ifNull: ['$register_time', null] }, { $literal: source.register_time ?? null }] }, { $literal: timestamp(source.register_time) === identity.register_time }, ] }; const valid = { $and: [ { $literal: saved.data_valid === true && calculatedLevel !== null }, currentIdentity, // Do not turn a concurrently established paid identity back into a never-payer. { $or: [ { $literal: saved.has_ever_paid === true }, { $and: [{ $ne: ['$vip_profile.has_ever_paid', true] }, { $lte: [{ $ifNull: ['$vip_level', 0] }, 0] }] }, ] }, ] }; const oldLevel = { $ifNull: ['$vip_level', { $literal: calculatedLevel }] }; const heldMetadata = (userField: string, value: any) => ({ $cond: [ { $ne: [{ $ifNull: ['$vip_level', null] }, null] }, { $ifNull: [`$vip_profile.${userField}`, null] }, { $literal: calculatedLevel === null ? null : value ?? null }, ] }); const metadata = { as_of: { $literal: saved.asOf }, updated_at: { $literal: saved.updatedAt }, data_valid: valid, identity: { $literal: identity }, has_ever_paid: { $or: [{ $eq: ['$vip_profile.has_ever_paid', true] }, { $gt: [{ $ifNull: ['$vip_level', 0] }, 0] }, { $literal: saved.has_ever_paid === true }] }, calculated_as_of: { $cond: [valid, { $literal: saved.profile_calculated_as_of ?? null }, heldMetadata('calculated_as_of', saved.profile_calculated_as_of)] }, rule_version: { $cond: [valid, { $literal: saved.profile_rule_version ?? null }, heldMetadata('rule_version', saved.profile_rule_version)] }, }; return { updateOne: { filter: { _id: saved._id }, upsert: false, // Merge only the two owned attributes; concurrent gameplay/account changes survive. update: [{ $replaceWith: { $cond: [ { $and: [ { $lte: [{ $ifNull: ['$vip_profile.as_of', 0] }, saved.asOf] }, { $lte: [{ $ifNull: ['$vip_profile.updated_at', 0] }, saved.updatedAt] }, ] }, { $mergeObjects: ['$$ROOT', { vip_level: { $cond: [valid, { $literal: calculatedLevel }, oldLevel] }, vip_profile: metadata }] }, '$$ROOT', ] } }], } }; })) : { modifiedCount: 0 }; processedUsers += batch.length; updatedUsers += result.modifiedCount + result.upsertedCount; updatedVipUsers += vipResult.modifiedCount; } const result = { asOf, updatedAt, scannedUsers, skippedNeverPaidUsers, processedUsers, updatedUsers, updatedVipUsers, validProfiles, invalidProfiles }; console.log('rechargeStats completed', result); return { code: 1, data: result, msg: '用户付费画像维护完成' }; }