206 lines
11 KiB
TypeScript
206 lines
11 KiB
TypeScript
import cloud from '@lafjs/cloud';
|
|
import { createHash, randomUUID } from 'crypto';
|
|
|
|
// Fixed identity: repeated manual calls resume this true-only repair, never start a new reset.
|
|
export const PAY_USER_REPAIR_ID = 'pay-user-true-only-v2-20260930';
|
|
const PAGE_SIZE = 100;
|
|
const RUN_BUDGET_MS = 15000;
|
|
const LEASE_MS = 60000;
|
|
const JOBS = 'payUserRepairJobs';
|
|
const AUDIT = 'payUserRepairAudit';
|
|
|
|
function auditId(id: any) {
|
|
return createHash('sha256').update(`${PAY_USER_REPAIR_ID}:${typeof id}:${String(id)}`).digest('hex');
|
|
}
|
|
|
|
function positiveInteger(value: any) {
|
|
return (typeof value === 'number' || typeof value === 'string')
|
|
&& Number.isSafeInteger(Number(value)) && Number(value) > 0;
|
|
}
|
|
|
|
// A missing order is not enough to override migration or historical payment evidence.
|
|
async function classify(mongo: any, user: any): Promise<{ paid?: boolean; reason: string }> {
|
|
if (typeof user.openid !== 'string' || !user.openid.trim()) return { reason: 'missing_openid' };
|
|
const linked = await mongo.collection('users').find({ openid: user.openid }, { projection: { _id: 1 } }).limit(2).toArray();
|
|
if (linked.length !== 1) return { reason: 'ambiguous_openid' };
|
|
let ambiguousOrder = false;
|
|
let pendingOrder = false;
|
|
const orders = mongo.collection('order').find({ openid: user.openid,
|
|
paymentAppEnv: { $ne: 'test' }, outTradeNo: { $not: /^wct_/ },
|
|
}, { projection: { state: 1, outTradeNo: 1, goodsPrice: 1, itemCount: 1 } });
|
|
try {
|
|
for await (const order of orders) {
|
|
if ((order.state === 1 || order.state === 2)
|
|
&& typeof order.outTradeNo === 'string' && !!order.outTradeNo.trim()
|
|
&& positiveInteger(order.goodsPrice) && positiveInteger(order.itemCount)
|
|
&& Number.isSafeInteger(Number(order.goodsPrice) * Number(order.itemCount))) {
|
|
return { paid: true, reason: 'confirmed_production_order' };
|
|
}
|
|
if (order.state === 0) pendingOrder = true;
|
|
else ambiguousOrder = true;
|
|
}
|
|
} finally { await orders.close(); }
|
|
// Some iOS paths mark the user before completing order confirmation. Never reset
|
|
// an account with a pending order; the revision check protects later callbacks.
|
|
if (pendingOrder) return { reason: 'pending_order' };
|
|
if (ambiguousOrder) return { reason: 'unverifiable_order' };
|
|
if (user.wucai_type !== undefined && ![-1, 3].includes(Number(user.wucai_type))) return { reason: 'migration_history_unavailable' };
|
|
const stats = await mongo.collection('userRechargeStats').findOne({ _id: user._id });
|
|
if (user.vip_level > 0 || user.vip_profile?.has_ever_paid === true
|
|
|| stats?.has_ever_paid === true || stats?.paid_status === 'paid'
|
|
|| stats?.amountTotal > 0 || stats?.orderCount > 0) return { reason: 'paid_history_missing' };
|
|
return { paid: false, reason: 'no_production_payment_evidence' };
|
|
}
|
|
|
|
function beforeImage(user: any) {
|
|
// Preserve field presence as well as BSON values; never back up session keys/tokens.
|
|
return Object.fromEntries(['pay_user', 'pay_user_revision', 'vip_level', 'vip_profile', 'pay_user_repair']
|
|
.map(field => [field, { exists: Object.prototype.hasOwnProperty.call(user, field), value: user[field] ?? null }]));
|
|
}
|
|
|
|
// Manually invoked from the cloud-function console, no public HTTP entry point. Deploy revision-aware payment
|
|
// writers BEFORE enabling this job. The job never modifies orders or grants rewards.
|
|
export default async function () {
|
|
if ((process.env.PAYMENT_APP_ENV || 'production') !== 'production') {
|
|
return { code: 0, msg: '仅允许正式环境执行付费身份修复' };
|
|
}
|
|
const mongo = cloud.mongo.db;
|
|
const users = mongo.collection('users');
|
|
const jobs = mongo.collection(JOBS);
|
|
const audit = mongo.collection(AUDIT);
|
|
const startedAt = Date.now();
|
|
const owner = randomUUID();
|
|
let job = await jobs.findOne({ _id: PAY_USER_REPAIR_ID });
|
|
if (job?.completedAt != null) return { code: 1, data: job, msg: '修复已完成，不再重复执行' };
|
|
if (!job) {
|
|
const last = await users.find({ pay_user: true }, { projection: { _id: 1 } }).sort({ _id: -1 }).limit(1).toArray();
|
|
try {
|
|
await jobs.updateOne({ _id: PAY_USER_REPAIR_ID }, { $setOnInsert: {
|
|
status: 'pending', createdAt: startedAt, upperId: last[0]?._id ?? null, afterId: null,
|
|
leaseUntil: 0, processed: 0, paid: 0, neverPaid: 0, skipped: 0,
|
|
} }, { upsert: true });
|
|
} catch (error: any) { if (error.code !== 11000) throw error; }
|
|
}
|
|
const claimed = await jobs.updateOne({ _id: PAY_USER_REPAIR_ID, completedAt: { $exists: false },
|
|
leaseUntil: { $lte: startedAt } }, { $set: { owner, leaseUntil: startedAt + LEASE_MS, status: 'running' } });
|
|
if (!claimed.matchedCount) return { code: 1, msg: '修复已完成或另一实例正在执行' };
|
|
const owned = () => ({ _id: PAY_USER_REPAIR_ID, owner, leaseUntil: { $gt: Date.now() } });
|
|
async function heartbeat() {
|
|
const result = await jobs.updateOne(owned(), { $set: { leaseUntil: Date.now() + LEASE_MS } });
|
|
if (!result.matchedCount) throw new Error('pay_user repair lease lost');
|
|
}
|
|
async function checkpoint(userId: any, result: any) {
|
|
const saved = await jobs.updateOne(owned(), { $set: { afterId: userId }, $inc: {
|
|
processed: 1, paid: result.paid === true ? 1 : 0,
|
|
neverPaid: result.paid === false ? 1 : 0, skipped: result.paid === undefined ? 1 : 0,
|
|
}, $unset: { pendingUserId: '' } });
|
|
if (!saved.matchedCount) throw new Error('pay_user repair lease lost at checkpoint');
|
|
job.afterId = userId;
|
|
delete job.pendingUserId;
|
|
}
|
|
try {
|
|
job = await jobs.findOne({ _id: PAY_USER_REPAIR_ID });
|
|
let finished = job.upperId === null;
|
|
let deferred = false;
|
|
const projection = { _id: 1, openid: 1, register_time: 1, wucai_type: 1, pay_user: 1,
|
|
pay_user_revision: 1, vip_level: 1, vip_profile: 1, pay_user_repair: 1 };
|
|
while (!finished && !deferred && Date.now() - startedAt < RUN_BUDGET_MS) {
|
|
// A repaired false user disappears from the true-only query. Recover its exact
|
|
// pending ID before paging, so a crash cannot lose the audit/count/checkpoint.
|
|
let page: any[];
|
|
if (job.pendingUserId != null) {
|
|
const pending = await users.findOne({ _id: job.pendingUserId }, { projection });
|
|
if (!pending) {
|
|
await heartbeat();
|
|
await audit.updateOne({ _id: auditId(job.pendingUserId) }, {
|
|
$setOnInsert: { batch: PAY_USER_REPAIR_ID, userId: job.pendingUserId, createdAt: Date.now() },
|
|
$set: { status: 'skipped', reason: 'user_deleted' },
|
|
}, { upsert: true });
|
|
await checkpoint(job.pendingUserId, {});
|
|
continue;
|
|
}
|
|
page = [pending];
|
|
} else {
|
|
page = await users.find({ pay_user: true, _id: { $lte: job.upperId,
|
|
...(job.afterId === null ? {} : { $gt: job.afterId }) } }, { projection })
|
|
.sort({ _id: 1 }).limit(PAGE_SIZE).toArray();
|
|
}
|
|
finished = page.length === 0;
|
|
for (const user of page) {
|
|
if (Date.now() - startedAt >= RUN_BUDGET_MS) break;
|
|
await heartbeat();
|
|
const tracked = await jobs.updateOne(owned(), { $set: { pendingUserId: user._id } });
|
|
if (!tracked.matchedCount) throw new Error('pay_user repair lease lost before user');
|
|
job.pendingUserId = user._id;
|
|
const id = auditId(user._id);
|
|
const record = await audit.findOne({ _id: id });
|
|
// The user marker closes the crash window between the user write and audit/checkpoint.
|
|
if (user.pay_user_repair?.batch === PAY_USER_REPAIR_ID) {
|
|
if (!record) throw new Error('pay_user repair backup missing');
|
|
await audit.updateOne({ _id: id }, { $set: { status: 'applied', appliedAt: user.pay_user_repair.at } });
|
|
await checkpoint(user._id, { paid: user.pay_user_repair.paid });
|
|
continue;
|
|
}
|
|
if (record?.status === 'skipped') {
|
|
await checkpoint(user._id, {});
|
|
continue;
|
|
}
|
|
// Only newly selected true users are recalculated. A pending user changed
|
|
// externally to false is recorded as skipped without touching its attributes.
|
|
const decision = user.pay_user === true ? await classify(mongo, user) : { reason: 'no_longer_marked_paid', paid: undefined };
|
|
await heartbeat();
|
|
await audit.updateOne({ _id: id }, { $setOnInsert: {
|
|
batch: PAY_USER_REPAIR_ID, userId: user._id, openid: user.openid ?? null,
|
|
createdAt: Date.now(), before: beforeImage(user), status: 'prepared',
|
|
} }, { upsert: true });
|
|
await audit.updateOne({ _id: id, status: { $nin: ['applied', 'skipped'] } }, {
|
|
$set: { status: decision.paid === undefined ? 'skipped' : 'prepared',
|
|
reason: decision.reason, plannedPaid: decision.paid ?? null },
|
|
});
|
|
if (decision.paid === undefined) {
|
|
await checkpoint(user._id, decision);
|
|
continue;
|
|
}
|
|
const now = Date.now();
|
|
// Compare both the payment token and user identity. A callback that already
|
|
// ran makes this fail; a later callback always writes true with another token.
|
|
const filter: any = { _id: user._id };
|
|
for (const field of ['openid', 'register_time', 'wucai_type', 'pay_user', 'pay_user_revision', 'vip_level', 'vip_profile']) {
|
|
filter[field] = Object.prototype.hasOwnProperty.call(user, field) ? { $eq: user[field] } : { $exists: false };
|
|
}
|
|
const changed = await users.updateOne(filter, { $set: {
|
|
pay_user: decision.paid, pay_user_revision: randomUUID(),
|
|
pay_user_repair: { batch: PAY_USER_REPAIR_ID, at: now, paid: decision.paid },
|
|
...(decision.paid ? {} : { vip_level: 0 }),
|
|
vip_profile: { ...(user.vip_profile || {}), data_valid: false,
|
|
has_ever_paid: decision.paid, updated_at: now,
|
|
...(decision.paid ? {} : { calculated_as_of: null, rule_version: null }) },
|
|
} });
|
|
if (!changed.matchedCount) {
|
|
// Do not advance past a concurrent payment/account change. Next manual call
|
|
// reads fresh identity and orders instead of applying a stale false result.
|
|
await audit.updateOne({ _id: id, status: { $ne: 'applied' } }, { $set: { status: 'deferred', reason: 'user_changed' } });
|
|
deferred = true;
|
|
break;
|
|
}
|
|
await audit.updateOne({ _id: id }, { $set: { status: 'applied', appliedAt: now } });
|
|
await checkpoint(user._id, decision);
|
|
}
|
|
}
|
|
const released = await jobs.updateOne(owned(), { $set: {
|
|
status: finished ? 'completed' : 'pending', leaseUntil: 0,
|
|
...(finished ? { completedAt: Date.now() } : {}),
|
|
}, $unset: { owner: '', lastError: '' } });
|
|
if (!released.matchedCount) throw new Error('pay_user repair lease lost at release');
|
|
job = await jobs.findOne({ _id: PAY_USER_REPAIR_ID });
|
|
console.log('repairPayUser progress', { status: job.status, processed: job.processed,
|
|
paid: job.paid, neverPaid: job.neverPaid, skipped: job.skipped });
|
|
return { code: 1, data: job, msg: finished ? '批量修复结束，异常用户见审计记录' : deferred ? '用户期间发生变更，请再次手动执行以续跑' : '已保存进度，请再次手动执行以继续剩余用户' };
|
|
} catch (error) {
|
|
await jobs.updateOne({ _id: PAY_USER_REPAIR_ID, owner }, {
|
|
$set: { status: 'pending', leaseUntil: 0, lastError: 'repair_failed', failedAt: Date.now() }, $unset: { owner: '' },
|
|
});
|
|
throw error;
|
|
}
|
|
}
|