xiaoshu-app/scripts/cloud/legacy-finance.js

// 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;index<specs.length;index++){
  const [account,table]=specs[index],all=[],seen=new Set();let total=-1;
  for(let page=1;page<=10000;page++){
   const response=await readSource({uid,stype:index+1,psize:1000,cpage:page}),rows=response.items;
   const count=Number(response.page.itemCount??rows.length),pages=Number(response.page.pageCount||Math.max(1,Math.ceil(count/1000)));
   if(!Number.isSafeInteger(count)||count<0||!Number.isSafeInteger(pages)||pages<1||page>pages||(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;index<all.length;index++){
    const raw=all[index];
    const id=String(raw.ExpHisID),delta=Number(raw.score??raw.Score),before=raw.score_before===null||raw.score_before===undefined||raw.score_before===''?null:Number(raw.score_before);
    const date=String(raw.HisTime||'').replace(' ','T'),iso=/([+-]\d{2}:?\d{2}|Z)$/.test(date)?date:date+'+08:00';
    if((raw.score??raw.Score)===undefined||(raw.score??raw.Score)===null||!Number.isFinite(delta)||(before!==null&&!Number.isFinite(before))||Number.isNaN(Date.parse(iso)))throw Error('finance_invalid_values');
    const hash=hashes[index].hash,prior=(priorById.get(id)||[]);
    if(prior.length>1)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;i<patches.length;i+=200)await t.none('UPDATE "'+table+'" l SET "legacyFinanceId"=x.id,"legacyFinanceHash"=x.hash,"legacyFinanceRaw"=x.raw FROM jsonb_to_recordset($1::jsonb) AS x(oid text,id text,hash text,raw jsonb) WHERE l."objectId"=x.oid',[JSON.stringify(patches.slice(i,i+200))]);
   for(let i=0;i<inserts.length;i+=200){const rows=inserts.slice(i,i+200),objectIds=await t.any('SELECT xs_object_id() id FROM generate_series(1,$1::int)',[rows.length]);if(objectIds.length!==rows.length)throw Error('finance_object_id_allocation_failed');for(let j=0;j<rows.length;j++)rows[j].objectId=objectIds[j].id;await t.none('INSERT INTO "'+table+'" SELECT * FROM jsonb_populate_recordset(NULL::"'+table+'",$1::jsonb)',[JSON.stringify(rows)]);}
   ids.sort((a,b)=>Number(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;
}