xiaoshu-app/scripts/cloud/legacy-direct-supplement.js

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