// 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 &&row.createdAt==null&&row.updatedAt==null&&!row.transactionId&&!row.legacyFinanceId&&!row.legacyFinanceHash &&Date.parse(row.hisTime)-Date.parse(iso)===8*60*60*1000 &&String(row.detail||'')===String(raw.Detail||'')&&String(row.remark||'')===String(raw.Remark||'') &&Number(row.operator||0)===(Number(raw.Operator)||0)&&Number(row.scoreType||0)===(Number(raw.ScoreType)||0); } async function importLegacyFinance(co,uid,runId,readSource){ if(!Number.isSafeInteger(uid)||uid<1||!/^[a-zA-Z0-9:_-]{8,100}$/.test(runId))throw Error('invalid_finance_identity'); const users=await Psql.query('SELECT "objectId",username,"legacyUserData","sourceKey","isDeleted","identityType" FROM "_User" WHERE company=$1 AND "legacyUserId"=$2',[co,uid]); const u=users[0];if(users.length!==1||u.isDeleted||u.identityType!=='member')throw Error('finance_member_identity_unverified'); const source=String(u.sourceKey||''),legacy=u.legacyUserData||{}; if(source.startsWith('reading-')||source.startsWith('reading:')||legacy.SyncExcluded===true||legacy.ReadingGeneratedLegacyId||(!source.startsWith('legacy')&&!source.startsWith('[["UserID",')))throw Error('finance_member_identity_unverified'); const attempt=runId+':'+Date.now(); const specs=[['Purse','UserExpDomP'],['SilverCoin','UserSIcon'],['UserExp','UserExpHis'],['UserPoint','UserUserPoint']],out={studentId:uid,accounts:[],created:0,conflicts:0}; for(let index=0;indexpages||(total>=0&&total!==count))throw Error('finance_source_shifted');total=count; // Durable private backup before projection. A retry gets a new attempt key to retain earlier evidence. await Psql.none('INSERT INTO "LegacyFinanceSourcePage"(company,"studentId",account,"runId",page,rows) VALUES($1,$2,$3,$4,$5,$6::jsonb) ON CONFLICT DO NOTHING',[co,uid,account,attempt,page,JSON.stringify(rows)]); for(const r of rows){const id=String(r.ExpHisID??'');if(!/^\d+$/.test(id)||Number(r.UserID)!==uid||seen.has(id)||String(r.UserName||'').toLowerCase()!==String(legacy.UserName||u.username).toLowerCase())throw Error('finance_source_identity_or_duplicates');seen.add(id);all.push(r);} if(page===pages)break;if(!rows.length||page===10000)throw Error('finance_incomplete_pages'); } if(all.length!==total)throw Error('finance_source_count_mismatch'); const result=await Psql.tx(async t=>{ await t.any("SELECT pg_advisory_xact_lock(hashtext('xs-credit:'||$1))",[co]); let created=0;const ids=[],patches=[],inserts=[]; const sourceIds=all.map(raw=>String(raw.ExpHisID)); const hashes=await t.any('SELECT ord::int idx,md5(value::text) hash FROM jsonb_array_elements($1::jsonb) WITH ORDINALITY AS e(value,ord)',[JSON.stringify(all)]); if(hashes.length!==all.length)throw Error('finance_hash_count_mismatch'); const existing=sourceIds.length?await t.any('SELECT to_jsonb(l) j FROM "'+table+'" l WHERE company=$1 AND ("legacyFinanceId"=ANY($2::text[]) OR "sourceKey"=ANY($3::text[]) OR to_jsonb(l)->>\'expHisId\'=ANY($2::text[]))',[co,sourceIds,sourceIds.map(id=>'[["ExpHisID",'+id+']]')]):[]; const priorById=new Map(),sourceIdSet=new Set(sourceIds); for(const {j} of existing){ const keys=[j.legacyFinanceId,j.expHisId,/^\[\["ExpHisID",(\d+)\]\]$/.exec(String(j.sourceKey||''))?.[1]].filter(Boolean).map(String); for(const key of new Set(keys))if(sourceIdSet.has(key)){const bucket=priorById.get(key)||[];bucket.push(j);priorById.set(key,bucket);} } for(let index=0;index1)throw Error('finance_duplicate_source'); if(prior.length){const j=prior[0];if(j.legacyFinanceId&&String(j.legacyFinanceId)!==id)throw Error('finance_existing_row_conflict');if(j.legacyFinanceHash&&j.legacyFinanceHash!==hash)throw Error('finance_source_conflict_preserved');if(Number(j.userId)!==uid||Number(j.score)!==delta||((j.scoreBefore===null?null:Number(j.scoreBefore))!==before))throw Error('finance_existing_row_conflict'); if(Date.parse(j.hisTime)!==Date.parse(iso)){ if(!untouchedLegacyFinanceTimezoneRow(j,raw,id,iso))throw Error('finance_existing_row_conflict'); const corrected=new Date(iso).toISOString(); await t.none('INSERT INTO "LegacySyncConflict"("objectId",company,"sourceKey",dataset,"legacyKey","conflictType",message,resolved,resolution,"createdAt","updatedAt") VALUES(xs_object_id(),$1,$2,\'finance\',$3,\'finance-legacy-timezone-repair\',$4,true,\'原始流水核验一致,仅修复早期导入时区偏移\',now(),now())',[co,'finance-timezone:'+account+':'+id,id,JSON.stringify({runId,account,studentId:uid,before:j,after:{...j,hisTime:corrected},sourceTime:raw.HisTime})]); await t.none('UPDATE "'+table+'" SET "hisTime"=$2::timestamptz WHERE "objectId"=$1',[j.objectId,corrected]); } if(j.legacyFinanceId!==id||j.legacyFinanceHash!==hash||!j.legacyFinanceRaw)patches.push({oid:j.objectId,id,hash,raw}); }else{ inserts.push({company:co,sourceKey:'legacy:finance:'+account+':'+id,userId:uid,score:delta,scoreBefore:before,hisTime:new Date(iso).toISOString(),createdAt:new Date().toISOString(),updatedAt:new Date().toISOString(),detail:String(raw.Detail||''),remark:String(raw.Remark||''),operator:Number(raw.Operator)||0,scoreType:Number(raw.ScoreType)||0,legacyFinanceId:id,legacyFinanceRaw:raw,legacyFinanceHash:hash});created++; }ids.push(id); } for(let i=0;iNumber(a)-Number(b)); const actual=await t.any('SELECT "legacyFinanceId" id,"legacyFinanceHash" hash FROM "'+table+'" WHERE company=$1 AND "userId"=$2 AND "legacyFinanceId"=ANY($3::text[]) ORDER BY "legacyFinanceId"::numeric',[co,uid,ids]); if(JSON.stringify(actual.map(r=>r.id))!==JSON.stringify(ids))throw Error('finance_id_reconciliation_failed'); const checksum=(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(actual)])).hash; await t.none('INSERT INTO "LegacyFinanceReceipt"(company,"studentId",account,"runId",ids,checksum) VALUES($1,$2,$3,$4,$5::jsonb,$6) ON CONFLICT(company,"studentId",account,"runId") DO UPDATE SET ids=EXCLUDED.ids,checksum=EXCLUDED.checksum',[co,uid,account,runId,JSON.stringify(ids),checksum]); if(account==='UserExp'){ const used=all.some(r=>Number(r.score??r.Score)!==0)||Number(legacy.UserExp||0)>0||Boolean(await t.oneOrNone('SELECT "objectId" FROM "LessonRecord" WHERE company=$1 AND xymz::text=$2 AND ("courseCategoryKey"=\'trial\' OR kclx::text=\'3\') LIMIT 1',[co,String(uid)])); await t.none('INSERT INTO "TrialEligibility"(company,"studentId",status,evidence) VALUES($1,$2,$3,$4::jsonb) ON CONFLICT(company,"studentId") DO UPDATE SET status=CASE WHEN "TrialEligibility".status=\'issued\' THEN \'issued\' ELSE EXCLUDED.status END,evidence="TrialEligibility".evidence||EXCLUDED.evidence,"updatedAt"=now()',[co,uid,used?'issued':'eligible',JSON.stringify({source:'WXAPP-full-trial-ledger',runId,count:all.length})]); } return{account,sourceCount:all.length,targetCount:actual.length,ids,checksum,created,reconciled:true}; }).catch(async error=>{await Psql.none('INSERT INTO "LegacySyncConflict"("objectId",company,"sourceKey",dataset,"legacyKey","conflictType",message,resolved,resolution,"createdAt","updatedAt") SELECT xs_object_id(),$1,$2,\'finance\',$3,\'finance-preserve-new\',$4,false,\'保留现有流水,待核对\',now(),now() WHERE NOT EXISTS(SELECT 1 FROM "LegacySyncConflict" WHERE company=$1 AND "sourceKey"=$2)',[co,'finance-conflict:'+uid+':'+account,String(uid),String(error.message)]);throw error;});out.accounts.push(result);out.created+=result.created; }return out; }