// 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; }