async function directSupplement(p,co){ async function all(action,args){const first=await directOld(action,{...args,psize:1000,cpage:1}),rows=[...first.items];for(let page=2;page<=Number(first.page.pageCount||1);page++)rows.push(...(await directOld(action,{...args,psize:1000,cpage:page})).items);return rows;} const users=await all('user_list',{uid:0}),nodes=await all('node_list',{nid:0}); const result={accounts:{source:users.length,created:0,matched:0,missing:0,conflicts:0,excluded:0,identityConflicts:0},nodes:{source:nodes.length,created:0,matched:0,missing:0},existingAccountBalancesUntouched:true}; 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]); 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;} if(decision==='identity-conflict'){ result.accounts.identityConflicts++;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,\'reading-legacy-id-collision\',\'阅读账号生成的编号不能代表旧系统身份\',false,\'保留原记录,待核实身份\',now(),now() WHERE NOT EXISTS(SELECT 1 FROM "LegacySyncConflict" WHERE company=$2 AND "sourceKey"=$1)',['direct-api:user-identity:'+id,co,String(id)]); continue; } if(!found.length){ await Psql.tx(async t=>{ // Serialize retries for the same legacy identity even if the first HTTP response was lost. await t.any('SELECT pg_advisory_xact_lock(hashtext($1))',['direct-sync-user:'+co+':'+id]); if(await t.oneOrNone('SELECT "objectId" FROM "_User" WHERE company=$1 AND "legacyUserId"=$2',[co,id]))return; if(![1,2,3].includes(value.body.legacyGroupId))throw Error('new_legacy_user_role_requires_confirmation'); let username=value.username; if(await Psql.oneOrNone('SELECT "objectId" FROM "_User" WHERE lower(username)=lower($1)',[username]))username='legacy_'+id+'_'+username; if(await Psql.oneOrNone('SELECT "objectId" FROM "_User" WHERE lower(username)=lower($1)',[username]))throw Error('new_legacy_user_login_conflict'); const tenant=await new Parse.Query('Company').get(co,{useMasterKey:true}); const user=new Parse.User(); for(const[k,v]of Object.entries(value.body))user.set(k,k==='company'?tenant:v?.__type==='Date'?new Date(v.iso):v); user.set('identityType',value.body.legacyGroupId===3?'coach':value.body.legacyGroupId===2?'store':'member');user.setUsername(username); user.setPassword((await Psql.one('SELECT gen_random_uuid()::text||gen_random_uuid()::text AS password')).password); await user.save(null,{useMasterKey:true}); await Psql.one('SELECT xs_enable_initial_legacy_login($1,$2)',[co,user.id]); const check=await Psql.query('SELECT "objectId" FROM "_User" WHERE company=$1 AND "legacyUserId"=$2',[co,id]);if(check.length!==1)throw Error('new_legacy_user_identity_verification_failed'); result.accounts.created++; }); }else{ const old=found[0].legacyUserData||{},fields=Object.keys(value.body.legacyUserData).filter(k=>!['LastLoginTime','LoginTimes'].includes(k)&&String(value.body.legacyUserData[k]??'')!==String(old[k]??'')); 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++; } 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\''); const knownNodes=new Set((await Psql.query('SELECT "nodeId" FROM "Node" WHERE company=$1',[co])).map(r=>Number(r.nodeId))); for(const row of nodes){const id=Number(row.NodeID);if(!Number.isSafeInteger(id)||id<1)throw Error('invalid_node_identity');if(knownNodes.has(id)){result.nodes.matched++;continue;} await Psql.tx(async t=>{await directLease(t,p,co);await t.any('SELECT pg_advisory_xact_lock(hashtext($1))',['direct-sync-node:'+co]);const found=await t.any('SELECT "objectId" FROM "Node" WHERE company=$1 AND "nodeId"=$2',[co,id]);if(found.length>1)throw Error('duplicate_node_identity');if(found.length)return; const body={objectId:(await t.one('SELECT substr(md5(gen_random_uuid()::text),1,20) id')).id,company:co,sourceKey:'legacy:node:'+id,createdAt:new Date().toISOString(),updatedAt:new Date().toISOString()}; for(const[field,key]of Object.entries(mapping)){const c=columns.find(c=>c.column_name===field);if(c&&row[key]!==undefined)body[field]=['numeric','double precision','integer'].includes(c.data_type)?Number(row[key]||0):c.data_type==='boolean'?row[key]===true||String(row[key])==='1':String(row[key]??'');} const fields=Object.keys(body).filter(k=>columns.some(c=>c.column_name===k));await t.none('INSERT INTO "Node" ('+fields.map(f=>'"'+f+'"').join(',')+') SELECT '+fields.map(f=>'v."'+f+'"').join(',')+' FROM jsonb_populate_record(NULL::"Node",$1::jsonb) v',[JSON.stringify(body)]);result.nodes.created++; });result.nodes.matched++; } return result; }