Explorar o código

fix(sync): reconcile legacy credit balances with local postings

彭峰 hai 8 horas
pai
achega
b4a428bd30

+ 45 - 0
docs/operations/legacy-credit-balance-reconciliation-2026-11-07.md

@@ -0,0 +1,45 @@
+# 历史课时余额对账修复 · 2026-11-07
+
+## 事件与根因
+
+客户反馈樊文昊高中(学员编号 1826)已完成 36 次训练,却仍有 19 次 60 分钟课时。
+
+正式数据核对:36 次训练包含 1 次体验课、35 次正式 60 分钟课。旧系统充值 35 次,至 10 月 30 日扣除 33 次、剩 2 次;新系统 11 月 1 日、6 日各正常扣 1 次,理论余量为 0。
+
+新系统此前保留了早期导入的 21 次余额。夜间同步导入了后续旧扣课流水和训练记录,但 `directSupplement` 为保护新系统修改而一律保留现有账户余额;`importLegacyFinance` 只对账原始流水,不更新余额。于是两次新结课把过期余额从 21 扣为 19,形成“记录已同步、余额未跟进”的差异。
+
+不能直接拿旧余额覆盖新余额,否则会丢失新系统的充值、退款、完课扣减和调整。
+
+## 修复规则
+
+- 分别保存旧系统来源余额、新系统本地流水及对账状态;首次修复必须有完整来源凭据、导入时间附近的可核对起点及连续的后续变动依据。
+- 正确余额为旧系统当前余额加新系统有效净变动。旧、新系统同一预约的扣课有独立凭据时只计一次,原始流水均保留;数量不匹配、多重关联等情况拒绝自动覆盖。
+- 历史 `UserPoint` 的部分扣减流水把扣后金额写在 `score_before`,只按这类已知源数据口径验证,不篡改原始流水。
+- 每次在公司课时事务锁与学员行锁内重新读取余额、核对全部来源 ID 和校验和,并核对本地流水;前后差额记入独立审计。重复执行零重复扣减。
+- 已建立基线后仅合并新增旧流水及本地变动;未知本地余额变化、来源缺页或凭据变化保留冲突。负余额保持真实值,不强行归零。体验余额增加仍需原有资格核验。
+- 夜间同步新增来源快照与余额对账回执,最终回执检查每位学员的余额对账覆盖,单列修正及冲突数量;未放开结算就绪。
+
+新增表:`LegacyCreditBalanceSnapshot`、`LegacyCreditBalanceState`、`LegacyCreditBalanceAudit`、`LegacyCreditBalanceReceipt`。现有课时、上课与消费流水不删除、不重放。
+
+## 检查与验收
+
+- 扫描 750 个旧系统身份及 551 个历史学员,共 2,204 个课时账户,读取 42,726 条课时流水。初步识别 111 位学员、230 个账户存在差额。
+- 同时识别 3 位学员、4 个预约存在新旧两端扣课凭据(含对应陪练时长);按同一预约计费一次合并。
+- 34 项相关测试通过,覆盖真实 PostgreSQL 事务、重跑幂等、原始流水保留、充值退款、负数与半小时、来源缺页、余额并发变化保护及重复扣课。
+- 11 个云函数定义校验通过。全量服务端执行 438 项:431 通过、7 失败。其中 `lesson-course-selection.test.mjs` 的两项失败在独立 `HEAD` 基线副本同样复现(测试夹具缺少 `sourceKey` 及后续课程状态错误);未跟踪的 `legacy-enrollment-reconciliation.test.mjs` 另有五项失败,其对应迁移文件尚不存在。本次未改动或提交这些文件,不宣称全量通过。
+- 正式数据库先进行事务回滚演练,再执行有审计的修复;原始来源用户余额在执行前重新读取,750 个身份余额与审计快照一致。
+- 樊文昊已修正 60 分钟课时和陪练时长各 19→0;正式 `app_credit_overview` 接口回读全部账户余额、预留和可用量均为 0。
+
+## 发布与证据
+
+后端选择性更新 `xiaoshu.sync.direct-v1` 的 `directSupplement`、`directControl`,新增余额计划与对账函数。其他线上函数逐项保留,源码哈希回读一致。未发布前端资源。
+
+专用修复批次:`credit-repair-20261107-v1`。私有备份、来源、计划、回滚演练及逐批回执:`~/.local/state/xiaoshu/cloud-releases/2026-11-06-credit-reconciliation/`。
+
+批量最终复核:551 位学员、2,204 个账户均符合逐笔计算的余额,修正 111 位学员、230 个账户,未决差额和对账冲突均为 0。42,726 条原始流水逐字段比较一致,樊文昊 36 条训练记录逐字段比较一致。3 位学员的 4 个跨系统重复预约均按一次扣课核算,原有两条训练记录和原始扣课凭据保留。
+
+针对樊文昊、3 位跨系统重复扣课学员及早期导入边界样本,共 5 位学员,在正式数据库事务回滚演练中重复执行两遍,新增修正均为 0、冲突为 0。修正累计数量从独立审计计算,重复执行不会把运行回执的历史修正数量清零。
+
+本次已验证实际修复和重复执行;下一次完整九组业务数据的夜间运行最终回执仍需届时验收。
+
+回退不能直接覆盖学员当前账户:先核对修复审计与随后新增流水,再撤销相应差额;代码可恢复发布前的直连函数备份。保留新增审计和业务凭据。

+ 10 - 4
scripts/cloud/legacy-direct-control.js

@@ -44,8 +44,11 @@ async function directControl(p,co,u) {
   const result=(await t.one('SELECT xs_legacy_review_income($1,$2,100,$3) result',[co,u.id,String(p.afterTaskId||'')])).result;
   await t.none('INSERT INTO "LegacyDirectBatch"(company,"runId",key,fingerprint,result) VALUES($1,$2,$3,$4,$5::jsonb)',[co,p.runId,p.batchKey,fingerprint,JSON.stringify({...result,modelId:0})]);return result;
  });
- if(p.action==='finance-members')return Psql.query('SELECT u."legacyUserId" id FROM "_User" u WHERE u.company=$1 AND u."identityType"=\'member\' AND COALESCE(u."isDeleted",false)=false AND (u."sourceKey" LIKE \'legacy%\' OR u."sourceKey" LIKE \'[["UserID",%\') AND COALESCE(u."legacyUserData"->>\'SyncExcluded\',\'false\')<>\'true\' AND (SELECT count(*) FROM "LegacyFinanceReceipt" f WHERE f.company=u.company AND f."studentId"=u."legacyUserId" AND f."runId"=$2)<4 ORDER BY u."legacyUserId"',[co,p.runId]);
- if(p.action==='finance')return importLegacyFinance(co,Number(p.studentId),p.runId,args=>directOld('user_point_list',args));
+ if(p.action==='finance-members')return Psql.query('SELECT u."legacyUserId" id FROM "_User" u WHERE u.company=$1 AND u."identityType"=\'member\' AND COALESCE(u."isDeleted",false)=false AND (u."sourceKey" LIKE \'legacy%\' OR u."sourceKey" LIKE \'[["UserID",%\') AND COALESCE(u."legacyUserData"->>\'SyncExcluded\',\'false\')<>\'true\' AND ((SELECT count(*) FROM "LegacyFinanceReceipt" f WHERE f.company=u.company AND f."studentId"=u."legacyUserId" AND f."runId"=$2)<4 OR NOT EXISTS(SELECT 1 FROM "LegacyCreditBalanceReceipt" b WHERE b.company=u.company AND b."studentId"=u."legacyUserId" AND b."runId"=$2)) ORDER BY u."legacyUserId"',[co,p.runId]);
+ if(p.action==='finance'){
+  const result=await importLegacyFinance(co,Number(p.studentId),p.runId,args=>directOld('user_point_list',args));
+  result.balances=await reconcileLegacyCreditBalances(co,Number(p.studentId),p.runId);return result;
+ }
  if(p.action==='source'){
   const m=Number(p.modelId);if(!specs[m])throw Error('invalid_model');
   if(p.id){if(!Number.isSafeInteger(p.id)||p.id<1)throw Error('invalid_id');return directOld('content_get',{id:p.id});}
@@ -101,11 +104,14 @@ async function directControl(p,co,u) {
   for(const [account,table] of [['Purse','UserExpDomP'],['SilverCoin','UserSIcon'],['UserExp','UserExpHis'],['UserPoint','UserUserPoint']]){
    const mismatch=await t.one('SELECT count(*)::int n FROM "LegacyFinanceReceipt" r WHERE r.company=$1 AND r."runId"=$2 AND r.account=$3 AND (jsonb_array_length(r.ids)<>(SELECT count(*) FROM "'+table+'" l WHERE l.company=r.company AND l."userId"=r."studentId" AND l."legacyFinanceId" IN (SELECT jsonb_array_elements_text(r.ids))) OR r.checksum<>(SELECT md5(COALESCE(jsonb_agg(jsonb_build_object(\'id\',l."legacyFinanceId",\'hash\',l."legacyFinanceHash") ORDER BY l."legacyFinanceId"::numeric),\'[]\'::jsonb)::text) FROM "'+table+'" l WHERE l.company=r.company AND l."userId"=r."studentId" AND l."legacyFinanceId" IN (SELECT jsonb_array_elements_text(r.ids))))',[co,p.runId,account]);if(mismatch.n)throw Error('finance_postimage_changed');
   }
-  const summary={...p.summary,finance:{members:members.length,streams:finance.length,rows:finance.reduce((n,f)=>n+f.ids.length,0),checksum:(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(finance.sort((a,b)=>(a.studentId+":"+a.account).localeCompare(b.studentId+":"+b.account)))])).hash,reconciled:true},createdBusinessRecords:d.reduce((n,x)=>n+x.created,0),refreshedBusinessRecords:d.reduce((n,x)=>n+x.updated,0),completedAt:new Date().toISOString(),source:'old-api/WXAPP',bridgeUsed:false,continuousSync:false,settlementReady:false,nightlyAutomation:true};
+  const balanceReceipts=await t.any('SELECT "studentId",result FROM "LegacyCreditBalanceReceipt" WHERE company=$1 AND "runId"=$2',[co,p.runId]);
+  if(members.some(m=>!balanceReceipts.some(b=>Number(b.studentId)===Number(m.id))))throw Error('balance_reconciliation_incomplete');
+  const balances={members:balanceReceipts.length,applied:balanceReceipts.reduce((n,b)=>n+Number(b.result.applied||0),0),conflicts:balanceReceipts.reduce((n,b)=>n+Number(b.result.conflicts||0),0)};
+  const summary={...p.summary,finance:{members:members.length,streams:finance.length,rows:finance.reduce((n,f)=>n+f.ids.length,0),checksum:(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(finance.sort((a,b)=>(a.studentId+":"+a.account).localeCompare(b.studentId+":"+b.account)))])).hash,reconciled:true,balances},createdBusinessRecords:d.reduce((n,x)=>n+x.created,0),refreshedBusinessRecords:d.reduce((n,x)=>n+x.updated,0),completedAt:new Date().toISOString(),source:'old-api/WXAPP',bridgeUsed:false,continuousSync:false,settlementReady:false,nightlyAutomation:true};
   const income=await t.one('SELECT COALESCE(sum((result->>\'earned\')::int),0)::int earned,COALESCE(sum((result->>\'excluded\')::int),0)::int excluded,COALESCE(sum((result->>\'amount\')::numeric),0) amount FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 AND result->>\'modelId\'=\'0\'',[co,p.runId]);
   const pending=await t.one('SELECT count(*)::int n FROM "LegacyReviewSyncState" WHERE company=$1 AND "pendingIncome"',[co]);
   summary.reviewIncome={...income,amount:Number(income.amount),pending:pending.n,errors:await t.any('SELECT \"taskId\",\"lastError\" AS reason FROM \"LegacyReviewSyncState\" WHERE company=$1 AND \"pendingIncome\" ORDER BY \"taskId\" LIMIT 25',[co])};
-  summary.conflicts=d.reduce((n,x)=>n+Number(x.conflicts||0),0)+Number(summary.accounts?.conflicts||0)+pending.n;
+  summary.conflicts=d.reduce((n,x)=>n+Number(x.conflicts||0),0)+Number(summary.accounts?.conflicts||0)+pending.n+balances.conflicts;
   summary.status=summary.conflicts?'completed-with-preserved-conflicts':'completed';
   await t.none('UPDATE "LegacyDirectRun" SET status=$3,summary=$4::jsonb,"finishedAt"=now(),"leaseUntil"=now() WHERE company=$1 AND "runId"=$2',[co,p.runId,summary.status,JSON.stringify(summary)]);
   await t.none('INSERT INTO "SysLog" ("objectId",company,"sourceKey",type1,type2,type3,detail,"createdAt","updatedAt") VALUES(substr(md5($1),1,20),$2,$1,\'migration\',\'legacy-direct-sync-completed\',\'nightly\',$3,now(),now())',['direct-nightly:'+p.runId,co,JSON.stringify({actor:u.id,action:'legacy-direct-sync-completed',after:summary})]);

+ 7 - 2
scripts/cloud/legacy-direct-supplement.js

@@ -5,7 +5,7 @@ async function directSupplement(p,co){
  for(const row of users){
   const value=projection(row),id=value.legacyUserId;if(!Number.isSafeInteger(id)||id<1)throw Error('invalid_legacy_user_identity');
   await directLease(Psql,p,co);
-  const found=await Psql.query('SELECT "objectId",username,"legacyUserData","isDeleted","sourceKey" FROM "_User" WHERE company=$1 AND "legacyUserId"=$2',[co,id]);
+  const found=await Psql.query('SELECT "objectId",username,"legacyUserData","isDeleted","sourceKey",to_jsonb(u)->>\'identityType\' AS "identityType" FROM "_User" u WHERE company=$1 AND "legacyUserId"=$2',[co,id]);
   if(found.length>1)throw Error('duplicate_legacy_user_identity');
   const decision=found.length?legacyAccountDecision(found[0]):'eligible';
   if(decision==='excluded'){result.accounts.excluded++;continue;}
@@ -38,7 +38,12 @@ async function directSupplement(p,co){
    if(fields.length){result.accounts.conflicts++;
     await Psql.none('INSERT INTO "LegacySyncConflict" ("objectId",company,"sourceKey",dataset,"legacyKey","conflictType",message,resolved,resolution,"createdAt","updatedAt") SELECT substr(md5($1),1,20),$2,$1,\'users\',$3,\'direct-api-preserve-new\',$4,false,\'保留新系统账号及余额\',now(),now() WHERE NOT EXISTS(SELECT 1 FROM "LegacySyncConflict" WHERE company=$2 AND "sourceKey"=$1)',['direct-api:user:'+id,co,String(id),JSON.stringify({fields,policy:'preserve-new'})]);
    }
-  }result.accounts.matched++;
+  }
+  if(found[0]?.identityType==='member'||!found.length&&value.body.legacyGroupId===1){
+   const balances=Object.fromEntries(['Purse','SilverCoin','UserExp','UserPoint'].map(account=>[account,value.body.legacyUserData[account]]));
+   await Psql.none('INSERT INTO "LegacyCreditBalanceSnapshot"(company,"studentId","runId",balances) VALUES($1,$2,$3,$4::jsonb) ON CONFLICT(company,"studentId","runId") DO UPDATE SET balances=EXCLUDED.balances,"capturedAt"=now()',[co,id,p.runId,JSON.stringify(balances)]);
+  }
+  result.accounts.matched++;
  }
  const mapping={nodeId:'NodeID',parentId:'ParentID',nodeName:'NodeName',depth:'Depth',orderId:'OrderID',contentModel:'ContentModel',zstatus:'ZStatus',nodeType:'NodeType',nodeDir:'NodeDir',nodePicUrl:'NodePic',description:'Description',cuser:'CUser',cuname:'CUName'};
  const columns=await Psql.query('SELECT column_name,data_type FROM information_schema.columns WHERE table_schema=\'public\' AND table_name=\'Node\'');

+ 79 - 0
scripts/cloud/legacy-finance.js

@@ -1,3 +1,82 @@
+// Merge source changes with local postings; never replace a live balance with a legacy snapshot.
+function legacyCreditBalancePlan(p) {
+ const conflict=reason=>({account:p.account,status:'conflict',reason});
+ const equal=(a,b)=>Math.abs(a-b)<0.000001,round=n=>Math.round(n*100)/100;
+ if(!['Purse','SilverCoin','UserExp','UserPoint'].includes(p.account)||![p.current,p.sourceBalance].every(v=>v!==null&&v!==''&&Number.isFinite(Number(v))))return conflict('invalid_balance');
+ const legacy=[...p.legacy].sort((a,b)=>Number(a.legacyFinanceId)-Number(b.legacyFinanceId)),ids=legacy.map(r=>String(r.legacyFinanceId));
+ if(new Set(ids).size!==ids.length||ids.some(id=>!/^\d+$/.test(id))||ids.length!==p.receiptIds.length||ids.some(id=>!p.receiptIds.map(String).includes(id)))return conflict('source_coverage');
+ if(legacy.some(r=>r.score==null||!Number.isFinite(Number(r.score))||!Number.isFinite(Date.parse(r.hisTime))))return conflict('invalid_source_row');
+ if(p.local.some(r=>!String(r.sourceKey||'').startsWith('credit:')||r.score==null||!Number.isFinite(Number(r.score))))return conflict('unclassified_local_row');
+ const localNet=round(p.local.reduce((sum,r)=>sum+Number(r.score),0)),sourceBalance=Number(p.sourceBalance),current=Number(p.current),previous=p.previous;
+ let changes;
+ if(previous){
+  if(previous.sourceIds.some(id=>!ids.includes(String(id))))return conflict('source_removed');
+  if(!equal(current,Number(previous.balance)+localNet-Number(previous.localNet)))return conflict('local_balance_conflict');
+  changes=legacy.filter(r=>!previous.sourceIds.map(String).includes(String(r.legacyFinanceId)));
+  if(!equal(sourceBalance,Number(previous.sourceBalance)+changes.reduce((sum,r)=>sum+Number(r.score),0)))return conflict('source_balance_conflict');
+ }else{
+  const importedAt=Date.parse(p.importedAt),migrationFloor=Date.parse('2026-09-01T00:00:00+08:00');
+  if(!Number.isFinite(importedAt))return conflict('missing_import_time');
+  const cutoff=Math.max(importedAt,migrationFloor);
+  changes=legacy.filter(r=>Date.parse(r.hisTime)>cutoff);
+  let baseline=sourceBalance-changes.reduce((sum,r)=>sum+Number(r.score),0);
+  // Early imports captured a source page before inserting users; tolerate that bounded capture window.
+  if(!equal(current,baseline+localNet)){
+   const window=legacy.filter(r=>Date.parse(r.hisTime)<=cutoff&&Date.parse(r.hisTime)>Math.max(importedAt-86400000,migrationFloor)).reverse();
+   for(const row of window){changes.unshift(row);baseline-=Number(row.score);if(equal(current,baseline+localNet))break;}
+  }
+  if(!equal(current,baseline+localNet))return conflict('balance_not_import_checkpoint');
+ }
+ let running=sourceBalance;
+ for(const row of [...changes].reverse()){
+  const after=running,before=running-Number(row.score);running=before;
+  // WXAPP's UserPoint debit rows historically put the post-debit amount in score_before.
+  if(row.scoreBefore!=null&&!equal(Number(row.scoreBefore),before)&&!(p.account==='UserPoint'&&Number(row.score)<0&&equal(Number(row.scoreBefore),after)))return conflict('source_chain_conflict');
+ }
+ const duplicates=p.duplicates||[],localIds=new Set(),sourceIds=new Set();let duplicateCredit=0;
+ for(const pair of duplicates){
+  const local=p.local.find(r=>r.objectId===pair.localId),source=legacy.find(r=>String(r.legacyFinanceId)===String(pair.sourceId));
+  if(!local||!source||Number(local.score)>=0||!equal(Number(local.score),Number(source.score))||localIds.has(pair.localId)||sourceIds.has(String(pair.sourceId)))return conflict('ambiguous_duplicate_charge');
+  localIds.add(pair.localId);sourceIds.add(String(pair.sourceId));duplicateCredit-=Number(local.score);
+ }
+ duplicateCredit=round(duplicateCredit);
+ const balance=round(sourceBalance+localNet+duplicateCredit);
+ return {account:p.account,status:'ready',balance,delta:round(balance-current),sourceBalance,localNet,sourceIds:ids,duplicateCredit,duplicates};
+}
+
+async function reconcileLegacyCreditBalances(co,uid,runId) {
+ return Psql.tx(async t=>{
+  await t.any("SELECT pg_advisory_xact_lock(hashtext('xs-credit:'||$1))",[co]);
+  const users=await t.any('SELECT "objectId","sourceKey","createdAt","legacyUserData","identityType","isDeleted" FROM "_User" WHERE company=$1 AND "legacyUserId"=$2 FOR UPDATE',[co,uid]);
+  const u=users[0],snapshot=await t.oneOrNone('SELECT balances FROM "LegacyCreditBalanceSnapshot" WHERE company=$1 AND "studentId"=$2 AND "runId"=$3',[co,uid,runId]);
+  if(users.length!==1||u.isDeleted||u.identityType!=='member'||!/^legacy|^\[\["UserID",/.test(u.sourceKey||'')||u.legacyUserData?.SyncExcluded||u.legacyUserData?.ReadingGeneratedLegacyId)throw Error('balance_member_identity_unverified');
+  if(!snapshot)return {studentId:uid,applied:0,conflicts:1,reason:'missing_source_snapshot',accounts:[]};
+  const accounts=[],next={...u.legacyUserData};let applied=0;
+  for(const [account,table] of [['Purse','UserExpDomP'],['SilverCoin','UserSIcon'],['UserExp','UserExpHis'],['UserPoint','UserUserPoint']]){
+   const receipt=await t.oneOrNone('SELECT ids,checksum FROM "LegacyFinanceReceipt" WHERE company=$1 AND "studentId"=$2 AND account=$3 AND "runId"=$4',[co,uid,account,runId]);
+   const rows=await t.any('SELECT "objectId","sourceKey",score,"scoreBefore","hisTime","legacyFinanceId","legacyFinanceHash" FROM "'+table+'" WHERE company=$1 AND "userId"=$2',[co,uid]);
+   const previous=await t.oneOrNone('SELECT balance,"sourceBalance","localNet","sourceIds" FROM "LegacyCreditBalanceState" WHERE company=$1 AND "studentId"=$2 AND account=$3',[co,uid,account]);
+   const legacy=rows.filter(r=>r.legacyFinanceId),local=rows.filter(r=>!r.legacyFinanceId);
+   const actual=await t.one('SELECT md5(COALESCE(jsonb_agg(jsonb_build_object(\'id\',"legacyFinanceId",\'hash\',"legacyFinanceHash") ORDER BY "legacyFinanceId"::numeric),\'[]\'::jsonb)::text) checksum FROM "'+table+'" WHERE company=$1 AND "userId"=$2 AND "legacyFinanceId" IS NOT NULL',[co,uid]);
+   const duplicates=await t.any('SELECT local."objectId" AS "localId",old."legacyFinanceId" AS "sourceId" FROM "'+table+'" local JOIN "CommonModel" c ON c.company=local.company AND c."modelId"=54 AND local."transactionId"=\'complete:\'||c."objectId" JOIN "'+table+'" old ON old.company=local.company AND old."userId"=local."userId" AND old."legacyFinanceId" IS NOT NULL AND old.score<0 AND old.detail=old."legacyFinanceRaw"->>\'Detail\' AND old.detail~(\'订单[::]\'||c."generalId"::text||\'([^0-9]|$)\') WHERE local.company=$1 AND local."userId"=$2 AND local.score<0 AND local."legacyFinanceId" IS NULL',[co,uid]);
+   const plan=!receipt||actual.checksum!==receipt.checksum?{account,status:'conflict',reason:'receipt_checksum'}:legacyCreditBalancePlan({account,current:Number(next[account]||0),sourceBalance:snapshot.balances[account],importedAt:u.createdAt,legacy,local,receiptIds:receipt.ids,previous,duplicates});
+   if(plan.status==='ready'&&account==='UserExp'&&plan.delta>0)Object.assign(plan,{status:'conflict',reason:'trial_increase_requires_confirmation'});
+   accounts.push(plan);
+   if(plan.status!=='ready')continue;
+   if(plan.delta){
+    await t.none('INSERT INTO "LegacyCreditBalanceAudit"(company,"studentId",account,"runId",before,after,delta,evidence) VALUES($1,$2,$3,$4,$5,$6,$7,$8::jsonb)',[co,uid,account,runId,Number(next[account]||0),plan.balance,plan.delta,JSON.stringify({sourceBalance:plan.sourceBalance,localNet:plan.localNet,sourceIds:plan.sourceIds,duplicateCredit:plan.duplicateCredit,duplicates:plan.duplicates,previous,importedAt:u.createdAt,receiptChecksum:receipt.checksum})]);
+    next[account]=plan.balance;applied++;
+   }
+   await t.none('INSERT INTO "LegacyCreditBalanceState"(company,"studentId",account,balance,"sourceBalance","localNet","sourceIds","runId") VALUES($1,$2,$3,$4,$5,$6,$7::jsonb,$8) ON CONFLICT(company,"studentId",account) DO UPDATE SET balance=EXCLUDED.balance,"sourceBalance"=EXCLUDED."sourceBalance","localNet"=EXCLUDED."localNet","sourceIds"=EXCLUDED."sourceIds","runId"=EXCLUDED."runId","updatedAt"=now()',[co,uid,account,plan.balance,plan.sourceBalance,plan.localNet,JSON.stringify(plan.sourceIds),runId]);
+  }
+  if(applied){await t.one("SELECT set_config('xiaoshu.credit_write','yes',true)");await t.none('UPDATE "_User" SET "legacyUserData"=$3::jsonb,"updatedAt"=now() WHERE company=$1 AND "objectId"=$2',[co,u.objectId,JSON.stringify(next)]);}
+  const result={studentId:uid,applied,conflicts:accounts.filter(p=>p.status==='conflict').length,accounts};
+  const cumulative=await t.one('SELECT count(*)::int n FROM "LegacyCreditBalanceAudit" WHERE company=$1 AND "studentId"=$2 AND "runId"=$3',[co,uid,runId]);
+  await t.none('INSERT INTO "LegacyCreditBalanceReceipt"(company,"studentId","runId",result) VALUES($1,$2,$3,$4::jsonb) ON CONFLICT(company,"studentId","runId") DO UPDATE SET result=EXCLUDED.result,"updatedAt"=now()',[co,uid,runId,JSON.stringify({...result,applied:cumulative.n})]);
+  return result;
+ });
+}
+
 // Server fetches authoritative WXAPP rows; clients cannot submit projected ledger data.
 function untouchedLegacyFinanceTimezoneRow(row,raw,id,iso){
  return row.sourceKey==='[["ExpHisID",'+id+']]'&&String(row.expHisId)===id

+ 1 - 1
scripts/deploy-direct-sync.mjs

@@ -16,7 +16,7 @@ const path='xiaoshu/system/direct-deploy-'+randomUUID(),expiry=Date.now()+600000
 const reviewSql=readFileSync(new URL('./sql/legacy-review-sync.sql',import.meta.url),'utf8').replace(/^BEGIN;\s*$/m,'').replace(/^COMMIT;\s*$/m,'');
 const atomicSql=readFileSync(new URL('./sql/mobile-teaching.sql',import.meta.url),'utf8').match(/CREATE OR REPLACE FUNCTION xs_mobile_teaching_write\([\s\S]*?\$\$;/)?.[0];
 if(!atomicSql)throw Error('Missing atomic teaching writer');
-const sql=[readFileSync(new URL('./sql/legacy-direct-sync.sql',import.meta.url),'utf8'),reviewSql,atomicSql].join('\n');let helperId;
+const sql=[readFileSync(new URL('./sql/legacy-direct-sync.sql',import.meta.url),'utf8'),readFileSync(new URL('./sql/legacy-credit-balances.sql',import.meta.url),'utf8'),reviewSql,atomicSql].join('\n');let helperId;
 try{
  const code=`async function handler(request,response){if(!request.user||request.user.id!==${JSON.stringify(session.user.objectId)}||Date.now()>${expiry})return response.status(403).json({success:false});try{await Psql.tx(async t=>t.none(${JSON.stringify(sql)}));return response.json({success:true,data:await Psql.query(${JSON.stringify("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema='public'")})});}catch(e){return response.status(500).json({success:false,message:e.message});}}`;
  new Function(code);helperId=(await api('/classes/Function','POST',{name:path,path,type:'standalone',enabled:true,code,params:[],paramList:[],respType:'json'})).objectId;

+ 23 - 0
scripts/sql/legacy-credit-balances.sql

@@ -0,0 +1,23 @@
+-- Keep legacy source balances separate from live accounts and all local credit postings.
+CREATE TABLE IF NOT EXISTS "LegacyCreditBalanceSnapshot" (
+ company text NOT NULL,"studentId" numeric NOT NULL,"runId" text NOT NULL,
+ balances jsonb NOT NULL,"capturedAt" timestamptz NOT NULL DEFAULT now(),
+ PRIMARY KEY(company,"studentId","runId")
+);
+CREATE TABLE IF NOT EXISTS "LegacyCreditBalanceState" (
+ company text NOT NULL,"studentId" numeric NOT NULL,account text NOT NULL,
+ balance numeric NOT NULL,"sourceBalance" numeric NOT NULL,"localNet" numeric NOT NULL,
+ "sourceIds" jsonb NOT NULL,"runId" text NOT NULL,"updatedAt" timestamptz NOT NULL DEFAULT now(),
+ PRIMARY KEY(company,"studentId",account)
+);
+CREATE TABLE IF NOT EXISTS "LegacyCreditBalanceAudit" (
+ company text NOT NULL,"studentId" numeric NOT NULL,account text NOT NULL,"runId" text NOT NULL,
+ "before" numeric NOT NULL,"after" numeric NOT NULL,delta numeric NOT NULL,evidence jsonb NOT NULL,
+ "createdAt" timestamptz NOT NULL DEFAULT now(),PRIMARY KEY(company,"studentId",account,"runId")
+);
+CREATE TABLE IF NOT EXISTS "LegacyCreditBalanceReceipt" (
+ company text NOT NULL,"studentId" numeric NOT NULL,"runId" text NOT NULL,
+ result jsonb NOT NULL,"updatedAt" timestamptz NOT NULL DEFAULT now(),
+ PRIMARY KEY(company,"studentId","runId")
+);
+REVOKE ALL ON "LegacyCreditBalanceSnapshot","LegacyCreditBalanceState","LegacyCreditBalanceAudit","LegacyCreditBalanceReceipt" FROM PUBLIC;

+ 4 - 0
scripts/tests/direct-nightly-sync.test.mjs

@@ -21,6 +21,7 @@ async function call(params,role='super-admin',company='co'){
 }
 before(async()=>{
  await db.exec(readFileSync(new URL('../sql/legacy-direct-sync.sql',import.meta.url),'utf8'));
+ await db.exec(readFileSync(new URL('../sql/legacy-credit-balances.sql',import.meta.url),'utf8'));
  await db.exec('CREATE TABLE "CommonModel" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,"generalId" numeric,"itemId" numeric,"modelId" numeric,"createdAt" timestamptz,"updatedAt" timestamptz,status numeric,"teachingManaged" boolean,title text)');
  for(const [m,s] of Object.entries(specs)){
   const fields=legacyAddonFields[m]||(m==='52'?['sy','yb','lj','yp']:['content','synopsis','mp4','ico','author','source']);
@@ -220,7 +221,10 @@ test('只有全部数据集对账后才能完成,统计来自写入回执,
   await db.exec(`UPDATE "UserSIcon" SET "legacyFinanceId"='1';UPDATE "LegacyFinanceReceipt" SET ids='["1"]',checksum='tampered' WHERE account='SilverCoin';`);
   assert.match((await call({action:'finish',summary})).message,/finance_postimage_changed/);
   await sql(`UPDATE "LegacyFinanceReceipt" SET checksum=md5('[{"id":"1","hash":"checksum-1"}]'::jsonb::text) WHERE account='SilverCoin'`);
+  assert.match((await call({action:'finish',summary})).message,/balance_reconciliation_incomplete/);
+  await sql(`INSERT INTO "LegacyCreditBalanceReceipt"(company,"studentId","runId",result) VALUES('co',999,$1,'{"applied":2,"conflicts":1}')`,[runId]);
   const r=await call({action:'finish',summary});assert.equal(r.success,true,r.message);assert.equal(r.data.finance.rows,1);assert.equal(r.data.finance.members,1);
+  assert.equal(r.data.finance.balances.applied,2);assert.equal(r.data.finance.balances.conflicts,1);
 
   assert.equal(r.data.createdBusinessRecords,2);assert.equal(r.data.refreshedBusinessRecords,1);
   assert.equal(r.data.bridgeUsed,false);assert.equal(r.data.settlementReady,false);

+ 100 - 0
scripts/tests/legacy-credit-balance.test.mjs

@@ -0,0 +1,100 @@
+import {test} from 'node:test';
+import assert from 'node:assert/strict';
+import {readFileSync} from 'node:fs';
+import vm from 'node:vm';
+import {PGlite} from '@electric-sql/pglite';
+
+const ctx=vm.createContext({Date});
+vm.runInContext(readFileSync(new URL('../cloud/legacy-finance.js',import.meta.url),'utf8'),ctx);
+const sourceRows=[
+ {legacyFinanceId:'1',score:35,scoreBefore:0,hisTime:'2026-08-04T08:00:00Z'},
+ {legacyFinanceId:'2',score:-14,scoreBefore:35,hisTime:'2026-08-31T08:00:00Z'},
+ {legacyFinanceId:'3',score:-19,scoreBefore:21,hisTime:'2026-10-30T08:00:00Z'},
+];
+const input=(extra={})=>({account:'SilverCoin',current:19,sourceBalance:2,importedAt:'2026-09-01T03:11:28Z',
+ legacy:sourceRows,local:[{sourceKey:'credit:complete:first',score:-1},{sourceKey:'credit:complete:second',score:-1}],
+ receiptIds:['1','2','3'],...extra});
+const plan=p=>{assert.equal(typeof ctx.legacyCreditBalancePlan,'function');return ctx.legacyCreditBalancePlan(p);};
+
+test('樊文昊旧余额21已扣两次变19,补齐旧系统漏同步19次后为0,保留两笔新扣课',()=>{
+ const result=plan(input());assert.equal(result.status,'ready');assert.equal(result.balance,0);assert.equal(result.localNet,-2);assert.equal(result.delta,-19);
+});
+test('同时保留新系统充值、退款与完课扣减,不直接覆盖成旧余额',()=>{
+ const result=plan(input({current:24,local:[{sourceKey:'credit:transfer:new',score:10},{sourceKey:'credit:refund:new',score:-3},{sourceKey:'credit:complete:new',score:-4}]}));
+ assert.equal(result.balance,5);assert.equal(result.localNet,3);
+});
+test('陪练时长保留旧系统扣减行将score_before存成扣后值的原始口径',()=>{
+ const result=plan(input({account:'UserPoint',legacy:sourceRows.map(r=>r.legacyFinanceId==='3'?{...r,scoreBefore:2}:r)}));assert.equal(result.balance,0);
+});
+test('来源余额与导入后的原始流水不一致时保留当前余额',()=>{
+ assert.equal(plan(input({sourceBalance:3})).status,'conflict');
+});
+test('未知本地流水、来源缺页、重复来源与无凭据的余额改动不能被自动覆盖',()=>{
+ for(const patch of [{local:[{sourceKey:'manual:unknown',score:-2}]},{receiptIds:['1','2']},{legacy:[...sourceRows,sourceRows[2]]},{current:18}])assert.equal(plan(input(patch)).status,'conflict');
+});
+test('小数课时与旧系统负余额保持真实数值,不能把欠费强制归零',()=>{
+ const result=plan(input({account:'UserPoint',current:0.5,sourceBalance:-1,legacy:[{legacyFinanceId:'1',score:2,scoreBefore:0,hisTime:'2026-08-31T08:00:00Z'},{legacyFinanceId:'2',score:-3,scoreBefore:-1,hisTime:'2026-10-30T08:00:00Z'}],receiptIds:['1','2'],local:[{sourceKey:'credit:complete:new',score:-1.5}]}));assert.equal(result.balance,-2.5);
+});
+test('重跑不重复扣减;建立基线后只合并新增旧流水与新增本地流水',()=>{
+ const previous={balance:0,sourceBalance:2,localNet:-2,sourceIds:['1','2','3']};
+ assert.equal(plan(input({current:0,previous})).delta,0);
+ const next=plan(input({current:-1,sourceBalance:1,previous,legacy:[...sourceRows,{legacyFinanceId:'4',score:-1,scoreBefore:2,hisTime:'2026-11-06T08:00:00Z'}],receiptIds:['1','2','3','4'],local:[...input().local,{sourceKey:'credit:complete:third',score:-1}]}));
+ assert.equal(next.balance,-2);assert.equal(next.delta,-1);
+ assert.equal(plan(input({current:1,previous})).status,'conflict');
+});
+test('既有来源凭据被删除,或来源变化没有新增流水时拒绝重算',()=>{
+ const previous={balance:0,sourceBalance:2,localNet:-2,sourceIds:['1','2','3']};
+ assert.equal(plan(input({current:0,previous,receiptIds:['1','2'],legacy:sourceRows.slice(0,2)})).status,'conflict');
+ assert.equal(plan(input({current:0,previous,sourceBalance:5})).status,'conflict');
+});
+
+test('同一预约在新旧系统各扣一次时只计一次,不删除任一原始流水',()=>{
+ const legacy=[...sourceRows.slice(0,2),{...sourceRows[2],score:-18},{legacyFinanceId:'4',score:-1,scoreBefore:3,hisTime:'2026-10-31T08:00:00Z'}];
+ const p=input({legacy,receiptIds:['1','2','3','4'],local:[{objectId:'local-1',sourceKey:'credit:complete:first',score:-1},{objectId:'local-2',sourceKey:'credit:complete:second',score:-1}],duplicates:[{localId:'local-1',sourceId:'4'}]});
+ const r=plan(p);assert.equal(r.balance,1);assert.equal(r.duplicateCredit,1);assert.equal(r.delta,-18);
+ assert.equal(plan({...p,current:1,previous:{...r}}).delta,0);
+ assert.equal(plan({...p,duplicates:[{localId:'local-1',sourceId:'3'}]}).status,'conflict');
+ assert.equal(plan({...p,duplicates:[...p.duplicates,...p.duplicates]}).status,'conflict');
+});
+
+test('早期来源抓取先于账号入库时,只接受一天窗口内可逐笔核清的起点',()=>{
+ const result=plan(input({current:38,sourceBalance:21,importedAt:'2026-10-12T23:00:00Z',legacy:[{legacyFinanceId:'1',score:40,scoreBefore:0,hisTime:'2026-10-11T12:00:00Z'},{legacyFinanceId:'2',score:-14,scoreBefore:40,hisTime:'2026-10-12T15:00:00Z'},{legacyFinanceId:'3',score:-5,scoreBefore:26,hisTime:'2026-10-30T15:00:00Z'}],local:[{sourceKey:'credit:complete:one',score:-2}]}));
+ assert.equal(result.balance,19);assert.equal(result.delta,-19);
+});
+
+test('数据库修复原子更新两类课时,保留业务流水,重复执行零扣减且未知改动受保护',async()=>{
+ const db=new PGlite();try{
+  await db.exec(`CREATE TABLE "_User"("objectId" text,company text,"legacyUserId" numeric,"sourceKey" text,"createdAt" timestamptz,"updatedAt" timestamptz,"legacyUserData" jsonb,"identityType" text,"isDeleted" boolean);
+   CREATE TABLE "LegacyFinanceReceipt"(company text,"studentId" numeric,account text,"runId" text,ids jsonb,checksum text);
+   CREATE TABLE "CommonModel"(company text,"objectId" text,"modelId" numeric,"generalId" numeric);
+   INSERT INTO "_User" VALUES('fan','co',1826,'legacy-sync:user:1826','2026-09-01T03:11:28Z',now(),'{"SilverCoin":19,"UserPoint":19,"Purse":0,"UserExp":0,"HoneyName":"樊文昊"}','member',false);`);
+  await db.exec(readFileSync(new URL('../sql/legacy-credit-balances.sql',import.meta.url),'utf8'));
+  const query=async(s,a=[])=>(await db.query(s,a)).rows,one=async(s,a)=>(await query(s,a))[0];
+  const Psql={any:query,query,one,oneOrNone:async(s,a)=>(await query(s,a))[0]||null,none:async(s,a)=>{assert.equal((await query(s,a)).length,0,'Psql.none must not receive rows');}};
+  Psql.tx=async f=>{await db.exec('BEGIN');try{const r=await f(Psql);await db.exec('COMMIT');return r;}catch(e){await db.exec('ROLLBACK');throw e;}};
+  const live=vm.createContext({Date,Psql});vm.runInContext(readFileSync(new URL('../cloud/legacy-finance.js',import.meta.url),'utf8'),live);
+  for(const [account,table] of [['Purse','UserExpDomP'],['SilverCoin','UserSIcon'],['UserExp','UserExpHis'],['UserPoint','UserUserPoint']]){
+   await db.exec(`CREATE TABLE "${table}"("objectId" text,company text,"userId" numeric,"sourceKey" text,score numeric,"scoreBefore" numeric,"hisTime" timestamptz,"legacyFinanceId" text,"legacyFinanceHash" text,"transactionId" text,detail text,"legacyFinanceRaw" jsonb)`);
+   if(account==='SilverCoin'||account==='UserPoint'){
+    for(const row of sourceRows)await query(`INSERT INTO "${table}"("objectId",company,"userId","sourceKey",score,"scoreBefore","hisTime","legacyFinanceId","legacyFinanceHash") VALUES($1,'co',1826,$2,$3,$4,$5,$1,'verified-'||$1)`,[row.legacyFinanceId,'legacy:finance:'+row.legacyFinanceId,row.score,account==='UserPoint'&&row.score<0?row.scoreBefore+row.score:row.scoreBefore,row.hisTime]);
+    await db.exec(`INSERT INTO "${table}"("objectId",company,"userId","sourceKey",score) VALUES('local-1','co',1826,'credit:complete:first',-1),('local-2','co',1826,'credit:complete:second',-1)`);
+   }
+   await query(`INSERT INTO "LegacyFinanceReceipt" SELECT 'co',1826,$1,'repair-0001',COALESCE(jsonb_agg("legacyFinanceId" ORDER BY "legacyFinanceId"::numeric),'[]'::jsonb),md5(COALESCE(jsonb_agg(jsonb_build_object('id',"legacyFinanceId",'hash',"legacyFinanceHash") ORDER BY "legacyFinanceId"::numeric),'[]'::jsonb)::text) FROM "${table}" WHERE "legacyFinanceId" IS NOT NULL`,[account]);
+  }
+  await db.exec(`INSERT INTO "LegacyCreditBalanceSnapshot"(company,"studentId","runId",balances) VALUES('co',1826,'repair-0001','{"SilverCoin":2,"UserPoint":2,"Purse":0,"UserExp":0}')`);
+  const result=await live.reconcileLegacyCreditBalances('co',1826,'repair-0001');assert.equal(result.applied,2);assert.equal(result.conflicts,0);
+  const balance=(await one('SELECT "legacyUserData" b FROM "_User"')).b;assert.equal(balance.SilverCoin,0);assert.equal(balance.UserPoint,0);assert.equal(balance.HoneyName,'樊文昊');
+  assert.equal((await live.reconcileLegacyCreditBalances('co',1826,'repair-0001')).applied,0);
+  assert.equal((await one('SELECT count(*)::int n FROM "LegacyCreditBalanceAudit"')).n,2);
+  assert.equal((await one('SELECT result FROM "LegacyCreditBalanceReceipt"')).result.applied,2);
+  assert.equal((await one('SELECT count(*)::int n FROM "UserSIcon"')).n,5);
+  await db.exec(`INSERT INTO "CommonModel" VALUES('co','appointment',54,102);
+   UPDATE "UserSIcon" SET "transactionId"='complete:appointment' WHERE "objectId"='local-1';
+   INSERT INTO "UserSIcon"("objectId",company,"userId","sourceKey",score,"scoreBefore","hisTime","legacyFinanceId","legacyFinanceHash",detail,"legacyFinanceRaw") VALUES('4','co',1826,'legacy:finance:4',-1,2,'2026-11-06T08:00:00Z','4','verified-4','学习结束,订单:102','{"Detail":"学习结束,订单:102"}');
+   UPDATE "LegacyCreditBalanceSnapshot" SET balances=balances||'{"SilverCoin":1}';`);
+  await db.exec(`UPDATE "LegacyFinanceReceipt" SET ids='["1","2","3","4"]',checksum=(SELECT md5(jsonb_agg(jsonb_build_object('id',"legacyFinanceId",'hash',"legacyFinanceHash") ORDER BY "legacyFinanceId"::numeric)::text) FROM "UserSIcon" WHERE "legacyFinanceId" IS NOT NULL) WHERE account='SilverCoin'`);
+  const deduped=await live.reconcileLegacyCreditBalances('co',1826,'repair-0001');assert.equal(deduped.accounts.find(a=>a.account==='SilverCoin').duplicateCredit,1);assert.equal((await one('SELECT "legacyUserData" b FROM "_User"')).b.SilverCoin,0);
+  await db.exec(`UPDATE "_User" SET "legacyUserData"="legacyUserData"||'{"SilverCoin":7}'`);
+  const protectedResult=await live.reconcileLegacyCreditBalances('co',1826,'repair-0001');assert.equal(protectedResult.conflicts,1);assert.equal((await one('SELECT "legacyUserData" b FROM "_User"')).b.SilverCoin,7);
+ }finally{await db.close();}
+});