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

// 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=[];
   for(const raw of all){
    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=(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(raw)])).hash;
    const prior=await t.any('SELECT to_jsonb(l) j FROM "'+table+'" l WHERE company=$1 AND ("legacyFinanceId"=$2 OR "sourceKey"=$3 OR to_jsonb(l)->>\'expHisId\'=$2)',[co,id,'[["ExpHisID",'+id+']]']);
    if(prior.length>1)throw Error('finance_duplicate_source');
    if(prior.length){const j=prior[0].j;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]);
     }
     await t.none('UPDATE "'+table+'" SET "legacyFinanceId"=$2,"legacyFinanceHash"=$3,"legacyFinanceRaw"=$4::jsonb WHERE "objectId"=$1',[j.objectId,id,hash,JSON.stringify(raw)]);
    }else{
     await t.none('INSERT INTO "'+table+'" SELECT * FROM jsonb_populate_record(NULL::"'+table+'",$1::jsonb)',[JSON.stringify({objectId:(await t.one('SELECT xs_object_id() id')).id,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);
   }
   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;
}