xiaoshu-app/scripts/run-nightly-direct-sync.mjs

#!/usr/bin/env node
import {readFile,writeFile,mkdir,rename,rm,readdir} from 'node:fs/promises';
import {homedir} from 'node:os';
import {join,resolve} from 'node:path';
import {randomUUID,randomBytes,createHash} from 'node:crypto';
import {fileURLToPath} from 'node:url';
import {specs,modelOrder} from './lib/legacy-direct-specs.mjs';
import {projectionFor,legacyAddonFields,signatureValue} from './lib/legacy-direct-projection.mjs';

export function changesFor(m,source,target,fields){
 const result={};for(const field of fields){const key=Object.keys(source).find(k=>k.toLowerCase()===field.toLowerCase()),raw=source[key];
  if(raw===undefined||m===59&&field==='pf'&&(raw===null||raw===''))continue;
  const normalize=v=>legacyAddonFields[m]?signatureValue(m,field,v):String(v??'').trim();
  if(normalize(raw)!==normalize(target[field]))result[field]=normalize(raw);
 }return result;
}
export function scheduledBusinessDate(now=new Date()){
 const format=d=>new Intl.DateTimeFormat('en-CA',{timeZone:'Asia/Shanghai',year:'numeric',month:'2-digit',day:'2-digit'}).format(d);
 const hour=Number(new Intl.DateTimeFormat('en-GB',{timeZone:'Asia/Shanghai',hour:'2-digit',hourCycle:'h23'}).format(now));
 return format(new Date(now.getTime()-(hour<23?86400000:0)));
}
// Keep durable reports; only expire raw page caches of completed runs after seven days.
export async function pruneCompletedCaches(root,now=Date.now()){
 for(const name of await readdir(root)){
  if(!/^\d{4}-\d{2}-\d{2}-direct-[a-f0-9-]{36}\.json$/.test(name))continue;
  let report;try{report=JSON.parse(await readFile(join(root,name),'utf8'));}catch{continue;}
  if(!['completed','completed-with-preserved-conflicts'].includes(report.status)||!/^([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})$/.test(report.runId||''))continue;
  const finished=Date.parse(report.finishedAt||report.completedAt);
  if(Number.isFinite(finished)&&finished<now-7*86400000)await rm(join(root,'direct-'+report.runId),{recursive:true,force:true});
 }
}
export async function runDirectSync(env=process.env,args=process.argv.slice(2)){
 const root=resolve(env.XIAOSHU_SYNC_REPORT_DIR||join(homedir(),'.local/state/xiaoshu/legacy-sync'));
 await mkdir(root,{recursive:true,mode:0o700});
 const report={source:'old-api/WXAPP',bridgeUsed:false,startedAt:new Date().toISOString(),status:'running',datasets:[],failed:0};
 const owner=randomUUID(),token=env.XIAOSHU_SYNC_OPERATOR_TOKEN,url=(env.XIAOSHU_FUNCTION_URL||'https://server.xiaoshu.pro/api/functions').replace(/\/$/,'')+'/xiaoshu/sync/direct-v1';
 const lock=join(root,'direct-nightly.lock');let locked=false,runId,runRoot;
 const save=async(path,value)=>{const tmp=path+'.tmp';await writeFile(tmp,JSON.stringify(value),{mode:0o600});await rename(tmp,path);};
 const load=async path=>{try{return JSON.parse(await readFile(path,'utf8'));}catch(e){if(e.code==='ENOENT')return null;throw e;}};
 async function call(params,retries=0){
  for(let n=0;;n++){try{const r=await fetch(url,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({token,params:{...params,runId,owner}}),signal:AbortSignal.timeout(90000)}),j=await r.json();
   if(!r.ok||!j.success)throw Error(j.message||'direct_gateway_http_'+r.status);return j.data;
  }catch(e){if(n>=retries)throw e;await new Promise(r=>setTimeout(r,1000*(n+1)));}}
 }
 try{
  if(!token)throw Error('missing_sync_operator_session');
  if(env.XIAOSHU_SYNC_OPERATOR_EXPIRES_AT&&Date.parse(env.XIAOSHU_SYNC_OPERATOR_EXPIRES_AT)<=Date.now())throw Error('expired_sync_operator_session');
  try{await mkdir(lock,{mode:0o700});locked=true;}catch(e){
   if(e.code!=='EEXIST')throw e;
   const prior=await load(join(lock,'owner.json'));let exited=false;
   if(Number.isSafeInteger(prior?.pid)&&prior.pid>0){try{process.kill(prior.pid,0);}catch(probe){if(probe.code==='ESRCH')exited=true;}}
   if(!exited)throw Error('another_run_or_stale_lock');
   await rm(lock,{recursive:true});await mkdir(lock,{mode:0o700});locked=true;
  }
  await save(join(lock,'owner.json'),{pid:process.pid,owner,startedAt:report.startedAt});
  await pruneCompletedCaches(root);
  const status=await call({action:'status'},2);if(status.source!=='old-api/WXAPP'||status.bridgeUsed!==false)throw Error('wrong_direct_gateway');
  if(args.includes('--check')){report.status='ready';return report;}
  const claim=await call({action:'claim',proposedRunId:randomUUID(),...(args.includes('--scheduled')?{targetBusinessDate:scheduledBusinessDate()}: {})});
  if(claim.alreadyCompleted){Object.assign(report,{status:'already_completed',businessDate:claim.businessDate,previous:claim.summary});return report;}
  runId=claim.runId;report.runId=runId;report.businessDate=claim.businessDate;runRoot=join(root,'direct-'+runId);await mkdir(runRoot,{recursive:true,mode:0o700});
  report.supplement=await call({action:'supplement'},2);
  if(Number(report.supplement.accounts?.identityConflicts||0)>0)throw Error('legacy_account_identity_conflict: resolve reading-generated IDs before syncing business records');
  const company={__type:'Pointer',className:'Company',objectId:env.XIAOSHU_SYNC_COMPANY_ID||'7pIbDBJmKx'},projection=projectionFor(company,specs);
  const physical=b=>Object.fromEntries(Object.entries(b).map(([k,v])=>[k,v?.__type==='Date'?v.iso:v?.__type==='Pointer'?v.objectId:v]));
  for(const m of modelOrder){
   const attemptPath=join(runRoot,`${m}-attempt.json`);let attempt=(await load(attemptPath))?.attempt||0;
   while(true){try{
   const s=specs[m],fields=legacyAddonFields[m]||(m===52?['sy','yb','lj','yp']:['content','synopsis','mp4','ico','author','source']);
   const sourcePage=async page=>{const path=join(runRoot,`${m}-source-${page}.json`),cached=await load(path);if(cached)return cached;const value=await call({action:'source',modelId:m,page},2);await save(path,value);return value;};
   const first=await sourcePage(1),total=Number(first.page.itemCount??first.items.length),pages=Math.max(1,Number(first.page.pageCount)||Math.ceil(total/1000));
   const allIds=new Set(),stats={modelId:m,label:s.label,source:total,created:0,createdAddon:0,updated:0,conflicts:0,failed:0,pages,reconciled:false},contentHash=createHash('sha256');
   for(let page=1;page<=pages;page++){
    const source=page===1?first:await sourcePage(page);contentHash.update(JSON.stringify(source.items));
    const ids=source.items.map(r=>Number(r.GeneralID??r.generalId));for(const id of ids){if(!Number.isSafeInteger(id)||id<1)throw Error('invalid_source_id');allIds.add(id);}
    const donePath=join(runRoot,`${m}-page-${page}-done.json`);let result=await load(donePath);
    if(!result){
     const planPath=join(runRoot,`${m}-page-${page}-plan.json`);let plan=await load(planPath);
     if(!plan){
      const target=await call({action:'read',modelId:m,ids},2),map=new Map();for(const t of target){if(map.has(Number(t.generalId)))throw Error('duplicate_target_id');map.set(Number(t.generalId),t);}
      plan={append:[],patch:[],conflicts:[]};const seen=new Set();
      for(let row of source.items){const id=Number(row.GeneralID??row.generalId);if(seen.has(id))continue;seen.add(id);const t=map.get(id);
       if(!t?.addon){let item=t?Number(t.common.itemId):id;
        if(m===52||m===2){const detail=(await call({action:'source',modelId:m,id},2)).items[0];if(detail){row={...row,...detail};if(!t)item=Number(detail.ID??detail.id??id);}}
        for(const f of fields){const k=Object.keys(row).find(k=>k.toLowerCase()===f.toLowerCase());if(k)row={...row,[f]:row[k]};}
        const common=physical(projection.commonBody(m,row,item)),addon=physical(projection.addonBody(m,row,item,id));
        if(!legacyAddonFields[m])for(const f of fields){const k=Object.keys(row).find(k=>k.toLowerCase()===f.toLowerCase());if(k)addon[f]=String(row[k]??'');}
        common.objectId=randomBytes(10).toString('hex');addon.objectId=randomBytes(10).toString('hex');addon.id=item;plan.append.push({generalId:id,common,addon,before:t||null});continue;
       }
       const values=changesFor(m,row,t.addon,fields);
       const metadata=['title','subtitle','inputer'].filter(f=>{const k=Object.keys(row).find(k=>k.toLowerCase()===f);return k!==undefined&&String(row[k]??'').trim()!==String(t.common[f]??'').trim();});
       if(metadata.length)plan.conflicts.push({modelId:m,generalId:id,fields:metadata,reason:'保留新系统元数据'});
       if(Object.keys(values).length)plan.patch.push({generalId:id,commonId:t.common.objectId,addonId:t.addon.objectId,commonVersion:t.common.updatedAt,addonVersion:t.addon.updatedAt,values,before:t});
      }await save(planPath,plan);
     }
     result={created:0,createdAddon:0,updated:0,conflicts:[...plan.conflicts]};
     for(const action of ['append','patch'])for(let i=0;i<plan[action].length;i+=50){const rows=plan[action].slice(i,i+50),r=await call({action,modelId:m,rows,batchKey:`${m}:r${attempt}:${page}:${action}:${i}`},2);
      result.created+=r.createdCommon;result.createdAddon+=r.createdAddon;result.updated+=r.updated;result.conflicts.push(...r.conflicts.map(c=>({...c,modelId:m})));
      const post=new Map(r.postimages.map(t=>[Number(t.generalId),t]));
      for(const x of rows){const after=post.get(x.generalId);if(after&&action==='patch'&&Object.keys(changesFor(m,x.values,after.addon,fields)).length)throw Error('post_write_field_mismatch');}
     }
     const dedup=new Map();for(const c of result.conflicts){const previous=dedup.get(c.generalId);dedup.set(c.generalId,previous?{...c,fields:[...new Set([...(previous.fields||[]),...(c.fields||[])])]}:c);}result.conflicts=[...dedup.values()];
     for(let i=0;i<result.conflicts.length;i+=100)await call({action:'conflicts',rows:result.conflicts.slice(i,i+100)},2);
     await save(donePath,result);
    }
    for(const key of ['created','createdAddon','updated'])stats[key]+=result[key];stats.conflicts+=result.conflicts.length;
    if(page%20===0||page===pages)console.log(JSON.stringify({dataset:s.label,pagesDone:page,pages,created:stats.created,updated:stats.updated,conflicts:stats.conflicts}));
   }
   if(allIds.size!==total)throw Error('source_shifted_restart_dataset');
   stats.sourceContentChecksum=contentHash.digest('hex');stats.sourceIdChecksum=createHash('md5').update([...allIds].sort((a,b)=>a-b).join('\n')).digest('hex');
   const evidence=await call({action:'evidence',modelId:m,conflicts:stats.conflicts},2);if(evidence.sourceChecksum!==stats.sourceIdChecksum||!evidence.reconciled)throw Error('source_target_checksum_mismatch');
   Object.assign(stats,evidence);report.datasets.push(stats);await save(join(runRoot,'progress.json'),report);break;
   }catch(e){if(!String(e.message).includes('source_shifted_restart_dataset')||attempt>=2)throw e;attempt++;await call({action:'reset-dataset',modelId:m});for(const file of await readdir(runRoot))if(file.startsWith(m+'-')&&file!==m+'-attempt.json')await rm(join(runRoot,file));await save(attemptPath,{attempt});console.log(JSON.stringify({modelId:m,restartForSourceChanges:attempt}));}
   }
  }
  report.createdBusinessRecords=report.datasets.reduce((n,d)=>n+d.created,0);report.refreshedBusinessRecords=report.datasets.reduce((n,d)=>n+d.updated,0);report.accounts=report.supplement.accounts;report.nodes=report.supplement.nodes;report.conflicts=report.datasets.reduce((n,d)=>n+d.conflicts,0)+Number(report.accounts.conflicts||0);
  report.status=report.conflicts?'completed-with-preserved-conflicts':'completed';report.allCapturedSourceIdsReconciled=true;
  Object.assign(report,await call({action:'finish',summary:report},2));
  return report;
 }catch(e){report.status='blocked';report.reason=String(e.message||'sync_failed').replace(/r:[A-Za-z0-9]+/g,'[redacted]').slice(0,240);report.failed=1;return report;}
 finally{report.finishedAt=new Date().toISOString();await save(join(root,`${report.businessDate||new Intl.DateTimeFormat('en-CA',{timeZone:'Asia/Shanghai',year:'numeric',month:'2-digit',day:'2-digit'}).format(new Date())}-direct-${runId||owner}.json`),report);if(locked)await rm(lock,{recursive:true});}
}
if(process.argv[1]&&resolve(process.argv[1])===fileURLToPath(import.meta.url)){
 const r=await runDirectSync();console.log(JSON.stringify({status:r.status,reason:r.reason||'',runId:r.runId,created:r.createdBusinessRecords,updated:r.refreshedBusinessRecords,conflicts:r.conflicts}));
 process.exitCode=['completed','completed-with-preserved-conflicts','already_completed','ready'].includes(r.status)?0:1;
}