296 lines
15 KiB
TypeScript
296 lines
15 KiB
TypeScript
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<string, {
|
|
stats: ReturnType<typeof emptyStats>;
|
|
seen: Map<string, { amount: number; paidAt: number | null }>;
|
|
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: '用户付费画像维护完成' };
|
|
}
|