xiaoshu-app/scripts/tests/direct-nightly-sync.test.mjs

import {test,before,after} from 'node:test';
import assert from 'node:assert/strict';
import {readFileSync} from 'node:fs';
import {mkdtemp,rm,readdir,mkdir,writeFile} from 'node:fs/promises';
import {tmpdir} from 'node:os';
import {join} from 'node:path';
import vm from 'node:vm';
import {PGlite} from '@electric-sql/pglite';
import {buildDirectBundle} from '../lib/legacy-direct-bundle.mjs';
import {specs} from '../lib/legacy-direct-specs.mjs';
import {legacyAddonFields} from '../lib/legacy-direct-projection.mjs';
import {runDirectSync,changesFor,scheduledBusinessDate,pruneCompletedCaches} from '../run-nightly-direct-sync.mjs';
const db=new PGlite();let context,runId;const owner='test-owner-0001';
const sql=async(s,p=[])=>(await db.query(s,p)).rows;
const Psql={query:sql,any:sql,one:async(s,p)=>(await sql(s,p))[0],oneOrNone:async(s,p)=>(await sql(s,p))[0]||null,none:async(s,p)=>{await sql(s,p);}};
Psql.tx=async f=>{await db.exec('BEGIN');try{const r=await f(Psql);await db.exec('COMMIT');return r;}catch(e){await db.exec('ROLLBACK');throw e;}};
async function call(params,role='super-admin',company='co'){
 let code=200,body;const response={status:n=>{code=n;return response;},json:j=>{body=j;return j;}};
 await context.handler({user:{id:'operator',fetch:async()=>{},get:k=>({adminRoleKey:role,company:{id:company}}[k])},body:{params:{runId,owner,...params}}},response);
 return {code,...body};
}
before(async()=>{
 await db.exec(readFileSync(new URL('../sql/legacy-direct-sync.sql',import.meta.url),'utf8'));
 await db.exec('CREATE TABLE "CommonModel" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,"generalId" numeric,"itemId" numeric,"modelId" numeric,"createdAt" timestamptz,"updatedAt" timestamptz,status numeric,"teachingManaged" boolean,title text)');
 for(const [m,s] of Object.entries(specs)){
  const fields=legacyAddonFields[m]||(m==='52'?['sy','yb','lj','yp']:['content','synopsis','mp4','ico','author','source']);
  await db.exec('CREATE TABLE "'+s.className+'" ("objectId" text PRIMARY KEY,company text,id numeric,"sourceKey" text,"createdAt" timestamptz,"updatedAt" timestamptz,"teachingManaged" boolean,'+fields.map(f=>'"'+f+'" text').join(',')+')');
 }
 await db.exec('ALTER TABLE "MemoryPracticeRecord" ADD "rewardTaskKey" text, ADD "rewardDuplicate" boolean');
 const rewardSource=readFileSync(new URL('../sql/review-rewards.sql',import.meta.url),'utf8');await db.exec(rewardSource.match(/CREATE OR REPLACE FUNCTION xiaoshu_review_key[\s\S]*?\$\$;/)[0]);
 await db.exec('CREATE TABLE "ReviewReward" ("objectId" text,company text,"taskId" text,amount numeric,status text)');
 await db.exec('CREATE TABLE "LegacySyncConflict" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,dataset text,"legacyKey" text,"conflictType" text,message text,resolved boolean,resolution text,"createdAt" timestamptz,"updatedAt" timestamptz)');
 const columns=await sql("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema='public'");
 context={Psql,Parse:{Config:{get:async()=>({get:k=>({legacyScheduleApiUrl:'https://legacy.invalid/api',legacyScheduleApiId:'test',legacyScheduleApiKey:'test'}[k])})}},fetch:async()=>({ok:true,json:async()=>({retcode:0,result:[{GeneralID:101,kcid:'5',xxcs:1}],page:{itemCount:1,pageCount:1}})}),AbortSignal,Date,console};
 vm.createContext(context);vm.runInContext(buildDirectBundle(columns,[],'co'),context);
 const r=await call({action:'claim',proposedRunId:'test-run-00001'});assert.equal(r.success,true,JSON.stringify(r));runId=r.data.runId;
});
after(()=>db.close());
test('非超级管理员、其他帐套与另一执行者不能访问或写入',async()=>{
 assert.equal((await call({action:'status'},'ops-manager')).code,403);
 assert.equal((await call({action:'status'},'super-admin','other')).code,403);
 assert.match((await call({action:'claim',owner:'another-owner-0001',proposedRunId:'another-run-00001'})).message,/another_sync_running/);
 assert.match((await call({action:'read',modelId:53,ids:[],owner:'wrong-owner-0001'})).message,/lease/);
});
let original;
test('主附表原子新增、批次重试复用结果，不允许同一批次编号换内容',async()=>{
 const row={generalId:101,common:{objectId:'common101',status:99,title:'旧记录'},addon:{objectId:'addon101',id:101,kcid:'5',xxcs:'1'}};
 const first=await call({action:'append',modelId:53,batchKey:'append-101',rows:[row]});assert.equal(first.success,true);assert.equal(first.data.createdCommon,1);original=first.data.postimages[0];
 const again=await call({action:'append',modelId:53,batchKey:'append-101',rows:[row]});assert.deepEqual(JSON.parse(JSON.stringify(first.data)),JSON.parse(JSON.stringify(again.data)));
 assert.equal((await sql('SELECT count(*)::int n FROM "PracticeRecord"'))[0].n,1);
 assert.match((await call({action:'append',modelId:53,batchKey:'append-101',rows:[{...row,generalId:102}]})).message,/batch_key_reused/);
 const bad={generalId:102,common:{objectId:'common102',unknown:'bad'},addon:{objectId:'addon102',id:102,kcid:'5'}};
 assert.equal((await call({action:'append',modelId:53,batchKey:'bad-102',rows:[bad]})).success,false);
 assert.equal((await sql('SELECT count(*)::int n FROM "PracticeRecord" WHERE id=102'))[0].n,0);
});
test('已保存投影可刷新，修改后的记录保留；微秒级版本变化也拒绝覆盖',async()=>{
 const patch=()=>({generalId:101,commonId:original.common.objectId,addonId:original.addon.objectId,commonVersion:original.common.updatedAt,addonVersion:original.addon.updatedAt,values:{xxcs:'2'}});
 let r=await call({action:'patch',modelId:53,batchKey:'patch-101',rows:[patch()]});assert.equal(r.data.updated,1);original=r.data.postimages[0];
 await sql('UPDATE "PracticeRecord" SET xxcs=\'新系统修改\' WHERE id=101');
 r=await call({action:'patch',modelId:53,batchKey:'preserve-101',rows:[patch()]});assert.equal(r.data.updated,0);assert.equal(r.data.conflicts.length,1);
 await sql('UPDATE "PracticeRecord" SET xxcs=\'2\',"updatedAt"=\'2026-10-12T10:00:00.000002Z\' WHERE id=101');
 const stale={...patch(),addonVersion:'2026-10-12T10:00:00.000001+00:00'};
 r=await call({action:'patch',modelId:53,batchKey:'microsecond-101',rows:[stale]});assert.equal(r.data.updated,0);
});
test('服务端捕获真实源编号并重新核对投影，缺主附表时不能生成对账凭据',async()=>{
 assert.equal((await call({action:'source',modelId:53,page:1})).success,true);
 const evidence=await call({action:'evidence',modelId:53,conflicts:1});assert.equal(evidence.success,true);assert.equal(evidence.data.sourceChecksum,evidence.data.targetChecksum);
 await sql('DELETE FROM "PracticeRecord" WHERE id=101');
 assert.match((await call({action:'evidence',modelId:53})).message,/projection_reconciliation_failed/);
 assert.match((await call({action:'finish',summary:{datasets:[]}})).message,/reconciliation_incomplete/);
});
test('冲突记录重复保存不增加重复行',async()=>{
 for(let i=0;i<2;i++)assert.equal((await call({action:'conflicts',rows:[{modelId:53,generalId:101,fields:['xxcs']}]})).success,true);
 assert.equal((await sql('SELECT count(*)::int n FROM "LegacySyncConflict"'))[0].n,1);
});
test('字段差异按业务类型规范化，旧空评分不抹掉新评分',()=>{
 assert.deepEqual(changesFor(53,{xxcs:'1'},{xxcs:1},['xxcs']),{});
 assert.deepEqual(changesFor(59,{pf:''},{pf:5},['pf']),{});
 assert.deepEqual(changesFor(60,{FXZT:'1'},{fxzt:'0'},['fxzt']),{fxzt:'1'});
});
test('每日运行入口只使用直接接口、按依赖顺序并写私有报告，未对账不报成功',async()=>{
 const root=await mkdtemp(join(tmpdir(),'direct-sync-test-')),realFetch=globalThis.fetch,order=[];
 try{
  globalThis.fetch=async(url,options)=>{assert.ok(url.endsWith('/xiaoshu/sync/direct-v1'));const p=JSON.parse(options.body).params;let data;
   if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false};
   else if(p.action==='claim')data={runId:'mock-run-000001',businessDate:'2026-10-12'};
   else if(p.action==='supplement')data={accounts:{missing:0},nodes:{missing:0}};
   else if(p.action==='source'){order.push(p.modelId);data={items:[],page:{itemCount:0,pageCount:1}};}
   else if(p.action==='read')data=[];
   else if(p.action==='evidence')data={sourceChecksum:'d41d8cd98f00b204e9800998ecf8427e',targetChecksum:'d41d8cd98f00b204e9800998ecf8427e',reconciled:true};
   else if(p.action==='finish')data=p.summary;
   return {ok:true,json:async()=>({success:true,data})};
  };
  const r=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});assert.equal(r.status,'completed');assert.ok(order.indexOf(56)<order.indexOf(60));assert.ok(order.indexOf(54)<order.indexOf(59));
  assert.ok((await readdir(root)).some(f=>f.endsWith('.json')));assert.ok(!(await readdir(root)).includes('direct-nightly.lock'));
 }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
});


test('有新系统计薪凭据的复习不回退、不重算工资',async()=>{
 const row={generalId:301,common:{objectId:'memory-common',status:99},addon:{objectId:'memory-addon',id:301,yhid:'20',xxjlid:'33',plid:'30',kywrq:'2026-10-12',fxzt:'0'}};
 const inserted=await call({action:'append',modelId:60,batchKey:'memory-301',rows:[row]});assert.equal(inserted.success,true,inserted.message);const t=inserted.data.postimages[0];
 await sql('INSERT INTO "ReviewReward" VALUES(\'earned-reward\',\'co\',\'memory-addon\',1,\'earned\')');
 const r=await call({action:'patch',modelId:60,batchKey:'memory-protected',rows:[{generalId:301,commonId:t.common.objectId,addonId:t.addon.objectId,commonVersion:t.common.updatedAt,addonVersion:t.addon.updatedAt,values:{fxzt:'1'}}]});
 assert.equal(r.success,true,r.message);assert.equal(r.data.updated,0);assert.equal(r.data.conflicts.length,1);
 assert.equal((await sql('SELECT amount FROM "ReviewReward" WHERE "objectId"=\'earned-reward\''))[0].amount,'1');
 assert.equal((await sql('SELECT fxzt FROM "MemoryPracticeRecord" WHERE id=301'))[0].fxzt,'0');
});
test('北京时间23点前补昨天、23点后执行今天，跨UTC日期仍正确',()=>{
 assert.equal(scheduledBusinessDate(new Date('2026-10-12T14:59:00Z')),'2026-10-11');
 assert.equal(scheduledBusinessDate(new Date('2026-10-12T15:00:00Z')),'2026-10-12');
 assert.equal(scheduledBusinessDate(new Date('2026-10-12T16:01:00Z')),'2026-10-12');
});


test('新增账号与栏目幂等补齐，已有账户余额和目录名称保持不变',async()=>{
 await db.exec('CREATE TABLE "_User" ("objectId" text PRIMARY KEY,company text,username text UNIQUE,"legacyUserId" numeric,"legacyUserData" jsonb);CREATE TABLE "Node" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,"nodeId" numeric,"parentId" numeric,"nodeName" text,"createdAt" timestamptz,"updatedAt" timestamptz)');
 await sql('INSERT INTO "_User" VALUES(\'existing-user\',\'co\',\'existing\',20,\'{"Purse":99}\')');
 await sql('INSERT INTO "Node" ("objectId",company,"nodeId","nodeName") VALUES(\'existing-node\',\'co\',8,\'新系统目录\')');
 await db.exec('ALTER TABLE "_User" ADD COLUMN "isDeleted" boolean, ADD COLUMN "sourceKey" text');
 const previous=context.fetch;let passwords=0;
 context.Parse.Query=class{constructor(className){assert.equal(className,'Company');}async get(id){return{id,className:'Company'};}};
 context.Parse.Object={createWithoutData:()=>{throw Error('Production SDK does not support two-argument createWithoutData');}};
 context.Parse.User=class{constructor(){this.fields={};}set(k,v){this.fields[k]=v;}setUsername(v){this.fields.username=v;}setPassword(v){assert.ok(v.length>60);passwords++;}async save(){const f=this.fields;assert.equal(f.identityType,'member');assert.equal(f.adminRoleKey,undefined);assert.equal(f.password,undefined);await sql('INSERT INTO "_User" ("objectId",company,username,"legacyUserId","legacyUserData") VALUES($1,$2,$3,$4,$5::jsonb)',['new-user',f.company.id,f.username,f.legacyUserId,JSON.stringify(f.legacyUserData)]);}};
 context.fetch=async url=>{const action=new URL(url).searchParams.get('action');return {ok:true,json:async()=>({retcode:0,page:{pageCount:1},result:action==='user_list'?[{UserID:20,UserName:'existing',GroupID:1,Purse:1},{UserID:4000,UserName:'new-member',GroupID:1,Purse:10,Password:'ignored-hash'}]:[{NodeID:8,NodeName:'旧目录'},{NodeID:9,ParentID:8,NodeName:'新目录'}]})};};
 try{let r=await call({action:'supplement'});assert.equal(r.success,true,r.message);assert.equal(r.data.accounts.created,1);assert.equal(r.data.nodes.created,1);
  r=await call({action:'supplement'});assert.equal(r.success,true,r.message);assert.equal(r.data.accounts.created,0);assert.equal(r.data.nodes.created,0);assert.equal(passwords,1);
  assert.equal((await sql('SELECT "legacyUserData" FROM "_User" WHERE "legacyUserId"=20'))[0].legacyUserData.Purse,99);
  assert.equal((await sql('SELECT "nodeName" FROM "Node" WHERE "nodeId"=8'))[0].nodeName,'新系统目录');
 }finally{context.fetch=previous;}
});


test('未取得服务器对账证据时不完成运行；已退出进程的目录锁可恢复',async()=>{
 const root=await mkdtemp(join(tmpdir(),'direct-sync-negative-')),realFetch=globalThis.fetch;let finished=false;
 try{
  await mkdir(join(root,'direct-nightly.lock'));
  await writeFile(join(root,'direct-nightly.lock/owner.json'),JSON.stringify({pid:2147483647}));
  globalThis.fetch=async(url,options)=>{const p=JSON.parse(options.body).params;let data;
   if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false};
   else if(p.action==='claim')data={runId:'failed-run-000001',businessDate:'2026-10-12'};
   else if(p.action==='supplement')data={accounts:{missing:0},nodes:{missing:0}};
   else if(p.action==='source')data={items:[],page:{itemCount:0,pageCount:1}};
   else if(p.action==='read')data=[];
   else if(p.action==='evidence')data={sourceChecksum:'d41d8cd98f00b204e9800998ecf8427e',reconciled:false};
   else if(p.action==='finish'){finished=true;data=p.summary;}
   return {ok:true,json:async()=>({success:true,data})};
  };
  const r=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});
  assert.equal(r.status,'blocked');assert.match(r.reason,/checksum/);assert.equal(finished,false);
  assert.ok(!(await readdir(root)).includes('direct-nightly.lock'));
 }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
});

test('删除标记禁止恢复，阅读数字编号冲突不会当作小树身份匹配', async()=>{
 await sql('INSERT INTO "_User" ("objectId",company,username,"legacyUserId","legacyUserData","isDeleted","sourceKey") VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7)', ['deleted-user','co','deleted',55,'{"Purse":88,"SyncExcluded":true}',true,'legacy-sync:user:55']);
 await sql('INSERT INTO "_User" ("objectId",company,username,"legacyUserId","legacyUserData","isDeleted","sourceKey") VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7)', ['reading-user','co','reading',1855,'{"Purse":77}',false,'reading-unified:user:reading-user']);
 const previous=context.fetch;
 context.fetch=async url=>({ok:true,json:async()=>({retcode:0,page:{pageCount:1},result:new URL(url).searchParams.get('action')==='user_list'?[{UserID:55,UserName:'deleted',GroupID:3,State:1},{UserID:1855,UserName:'real-student',GroupID:1}]:[]})});
 try {
  const r=await call({action:'supplement'});assert.equal(r.success,true,r.message);
  assert.equal(r.data.accounts.excluded,1);assert.equal(r.data.accounts.identityConflicts,1);assert.equal(r.data.accounts.matched,0);assert.equal(r.data.accounts.created,0);
  assert.equal((await sql('SELECT "isDeleted" FROM "_User" WHERE "objectId"=$1',['deleted-user']))[0].isDeleted,true);
  assert.equal((await sql('SELECT "legacyUserData" FROM "_User" WHERE "objectId"=$1',['reading-user']))[0].legacyUserData.Purse,77);
  assert.equal((await sql('SELECT count(*)::int n FROM "LegacySyncConflict" WHERE "conflictType"=$1',['reading-legacy-id-collision']))[0].n,1);
 } finally {context.fetch=previous;}
});

test('只有全部数据集对账后才能完成，统计来自写入回执，完成重试幂等',async()=>{
 await db.exec('CREATE TABLE "SysLog" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,type1 text,type2 text,type3 text,detail text,"createdAt" timestamptz,"updatedAt" timestamptz)');
 const previous=context.fetch;
 context.fetch=async()=>({ok:true,json:async()=>({retcode:0,result:[],page:{itemCount:0,pageCount:1}})});
 try{
  const datasets=[];
  for(const modelId of [2,52,58,56,53,60,54,59,61]){
   assert.equal((await call({action:'source',modelId,page:1})).success,true);
   const e=await call({action:'evidence',modelId});assert.equal(e.success,true,e.message);
   datasets.push({modelId,created:999,updated:999,failed:0,reconciled:true});
  }
  const summary={status:'completed-with-preserved-conflicts',datasets,conflicts:1};
  const r=await call({action:'finish',summary});assert.equal(r.success,true,r.message);
  assert.equal(r.data.createdBusinessRecords,2);assert.equal(r.data.refreshedBusinessRecords,1);
  assert.equal(r.data.bridgeUsed,false);assert.equal(r.data.settlementReady,false);
  const again=await call({action:'finish',summary});assert.deepEqual(JSON.parse(JSON.stringify(again.data)),JSON.parse(JSON.stringify(r.data)));
  assert.equal((await sql('SELECT count(*)::int n FROM "SysLog"'))[0].n,1);
  assert.equal((await call({action:'claim',proposedRunId:'next-run-0000001'})).data.alreadyCompleted,true);
 }finally{context.fetch=previous;}
});


test('只清理七天前已完成运行的原始缓存，保留失败断点及报告',async()=>{
 const root=await mkdtemp(join(tmpdir(),'direct-sync-retention-'));
 const ids=['11111111-1111-1111-1111-111111111111','22222222-2222-2222-2222-222222222222','33333333-3333-3333-3333-333333333333'];
 try{
  for(let i=0;i<ids.length;i++){
   await mkdir(join(root,'direct-'+ids[i]));
   await writeFile(join(root,'2026-10-01-direct-'+ids[i]+'.json'),JSON.stringify({runId:ids[i],status:i===1?'blocked':'completed',finishedAt:i===2?'2026-10-12T12:00:00Z':'2026-10-01T12:00:00Z'}));
  }
  await pruneCompletedCaches(root,Date.parse('2026-10-13T00:00:00Z'));
  const files=await readdir(root);
  assert.ok(!files.includes('direct-'+ids[0]));assert.ok(files.includes('direct-'+ids[1]));assert.ok(files.includes('direct-'+ids[2]));
  assert.equal(files.filter(x=>x.endsWith('.json')).length,3);
 }finally{await rm(root,{recursive:true,force:true});}
});


test('账号编号冲突时停止业务数据同步，保留阻塞报告',async()=>{
 const root=await mkdtemp(join(tmpdir(),'direct-identity-block-')),realFetch=globalThis.fetch;let sourceCalls=0;
 try{
  globalThis.fetch=async(url,options)=>{const p=JSON.parse(options.body).params;let data;
   if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false};
   else if(p.action==='claim')data={runId:'identity-block-000001',businessDate:'2026-10-13'};
   else if(p.action==='supplement')data={accounts:{identityConflicts:1}};
   else {sourceCalls++;throw Error('business writes must not run');}
   return {ok:true,json:async()=>({success:true,data})};
  };
  const r=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});
  assert.equal(r.status,'blocked');assert.equal(sourceCalls,0);assert.match(JSON.stringify(r),/legacy_account_identity_conflict/);
 }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
});