xiaoshu-app/scripts/tests/payroll-read-snapshot.test.mjs

import {test,before,after} from 'node:test';import assert from 'node:assert/strict';import fs from 'node:fs';import vm from 'node:vm';import {PGlite} from '@electric-sql/pglite';
const db=new PGlite(),query=async(s,a=[])=>(await db.query(s,a)).rows,one=async(s,a=[])=>(await query(s,a))[0];
const read=p=>fs.readFileSync(new URL(p,import.meta.url),'utf8'),integrity=read('../sql/payroll-integrity.sql'),gateway=read('../deploy-admin-functions.mjs').match(/const operationsGatewayCode = String.raw`([\s\S]*?)\n`;/)[1],cloud=read('../cloud/payroll-integrity.js');
const fn=(s,n)=>s.match(new RegExp('CREATE OR REPLACE FUNCTION '+n+'\\([\\s\\S]*?\\$\\$;'))[0];
const fixedDate=class extends Date{constructor(v){super(v??'2026-11-10T00:00:00Z');}static now(){return new Date('2026-11-10T00:00:00Z').getTime();}};
let raw=[];
before(async()=>{
 await db.exec(`CREATE TABLE raw_sources(ord int,s jsonb);CREATE TABLE test_conflicts(s jsonb);
 CREATE TABLE "TeachingCompensationRule"("objectId" text,company text,revision int);
 CREATE TABLE "CourseTeachingProfile"("objectId" text,company text);
 CREATE TABLE "CoachPayRate"("objectId" text,company text);
 CREATE TABLE "ReviewRewardIncome"("objectId" text,company text,status text,"incomeMonth" text,amount numeric);
 CREATE TABLE "LegacyDirectRun"(company text,"runId" text,status text,"finishedAt" timestamptz);
 CREATE TABLE "LegacyDirectEvidence"(company text,"runId" text,model int,ids jsonb,"sourceCount" int,"sourceChecksum" text,"targetChecksum" text);
 CREATE TABLE "PayrollCutoffApproval"(company text,month text,version text);
 CREATE TABLE "CommonModel"(company text,"generalId" numeric,"modelId" numeric,"itemId" numeric);
 CREATE TABLE "CourseAppointment"(company text,id numeric);CREATE TABLE "LessonRecord"(company text,id numeric);CREATE TABLE "MemoryPracticeRecord"(company text,id numeric);
 CREATE TABLE "PayrollLine"(company text,"standardLessonKey" text,"batchId" text,excluded boolean,kind text);
 CREATE TABLE "_User"("objectId" text,company text,"legacyUserId" numeric,"legacyUserData" jsonb,"realName" text,nickname text,username text,mobile text);
 CREATE TABLE "PayrollReadAuthority"(name text PRIMARY KEY,hash text);
 CREATE FUNCTION xs_payroll_sources(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT s FROM raw_sources ORDER BY ord $$;
 CREATE FUNCTION xs_payroll_sources_bulk(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT s FROM raw_sources ORDER BY ord $$;
 CREATE FUNCTION xs_payroll_sources_cached(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT s FROM raw_sources ORDER BY ord $$;
 CREATE FUNCTION xs_payroll_cutoff_snapshot(co text,mon text) RETURNS jsonb LANGUAGE sql STABLE AS $$ SELECT '{}'::jsonb $$;
 CREATE FUNCTION xs_payroll_data_version(co text) RETURNS text LANGUAGE sql STABLE AS $$ SELECT md5(COALESCE(string_agg(s::text,'|' ORDER BY ord),'')) FROM raw_sources $$;
 CREATE FUNCTION xs_payroll_cutoff_conflicts(co text,mon text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT s FROM test_conflicts $$;`);
 await db.exec(fn(integrity,'xs_payroll_time'));await db.exec(fn(integrity,'xs_payroll_source_version'));
 await db.exec(`CREATE FUNCTION xs_payroll_source_version_fast(co text) RETURNS text LANGUAGE sql STABLE AS $$ SELECT xs_payroll_source_version(co) $$;`);
 await db.exec(fn(read('../sql/payroll-incremental-reads.sql'),'xs_payroll_authority_ready'));
 await db.exec(`INSERT INTO "PayrollReadAuthority" SELECT name,md5(pg_get_functiondef(to_regprocedure(name))) FROM (VALUES ('xs_payroll_sources(text)'),('xs_payroll_source_version(text)'),('xs_payroll_cutoff_snapshot(text,text)')) v(name);
 INSERT INTO "_User" VALUES('coach','co',30,'{"ParentUserID":10,"HoneyName":"旧姓名"}',NULL,NULL,'coach13812345678',''),('student','co',20,'{}','学生',NULL,NULL,NULL),('store','co',10,'{}','现门店',NULL,NULL,NULL),('store-old','co',11,'{}','历史门店',NULL,NULL,NULL);
 INSERT INTO "TeachingCompensationRule" VALUES('rule','co',1);INSERT INTO "CourseTeachingProfile" VALUES('profile','co');INSERT INTO "CoachPayRate" VALUES('rate','co');`);
 await db.exec(read('../sql/payroll-cutoff-fast.sql'));
 await db.exec(fn(read('../sql/payroll-cutoff-bulk.sql'),'xs_payroll_evidence_matches'));
 // This suite injects conflict results to isolate coherent snapshot semantics.
 // The set-based production conflict implementation has its own SQL tests.
 await db.exec(`CREATE FUNCTION xs_payroll_cutoff_conflicts_bulk(co text,mon text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT s FROM xs_payroll_cutoff_conflicts(co,mon) s $$;`);
 await db.exec(`CREATE FUNCTION xs_payroll_source_entries_scoped(co text,mon text,coach_id text,appointment_ids text[],with_text boolean)
 RETURNS TABLE(row_data jsonb,row_text text,ordinal bigint,standard_key text,month_candidate_key text,scope_data jsonb) LANGUAGE sql STABLE AS $$
 SELECT CASE WHEN EXISTS(SELECT 1 FROM raw_sources GROUP BY s->>'standardLessonKey' HAVING count(*)>1)
 OR CASE WHEN s->>'lessonAt' ~ '^\\d{4}-[0-9]{2}-[0-9]{2}T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](\\.[0-9]{1,3})?(Z|[+-][0-9]{2}:[0-9]{2})$'
 AND xs_payroll_time(s->>'lessonAt') IS NOT NULL THEN to_char(xs_payroll_time(s->>'lessonAt') AT TIME ZONE 'Asia/Shanghai','YYYY-MM')=mon ELSE true END THEN s END,
 CASE WHEN with_text THEN s::text END,ord::bigint,s->>'standardLessonKey',NULL::text,jsonb_build_array(s->'coachId',s->'jsmz',s->'storeId',s->'szmdid') FROM raw_sources $$;`);
 await db.exec(read('../sql/payroll-read-snapshot.sql'));
});
after(()=>db.close());
async function setRows(values){raw=values;await db.exec('DELETE FROM raw_sources');for(let i=0;i<values.length;i++)await query('INSERT INTO raw_sources VALUES($1,$2::jsonb)',[i,JSON.stringify(values[i])]);}
const lesson=(id,at,extra={})=>({objectId:id,lessonObjectId:id,standardLessonKey:'lesson:'+id,jsmz:'30',xymz:'20',kclx:'2',lessonAt:at,courseCategoryKey:'word',teacherPaySnapshot:40,...extra});
const snapshot=async(dashboard=false)=>(await one('SELECT xs_payroll_read_snapshot($1,$2,$3,true) AS r',['co','2026-11',dashboard])).r;
const plain=v=>JSON.parse(JSON.stringify(v));
function ctx(accelerated=true){const calls=[],psql={query,one,oneOrNone:one,tx:async run=>{calls.push('tx');return db.transaction(async tx=>run({none:async sql=>{calls.push(sql);await tx.exec(sql);},one:async(s,a)=>(await tx.query(s,a)).rows[0]}));}};const c=vm.createContext({Date:fixedDate,Psql:psql});vm.runInContext(gateway+'\n'+cloud,c);c.companyIdOf=()=> 'co';c.payRateRows=async()=>[];c.teachingRuleRows=async()=>[];c.courseTeachingRows=async()=>[];if(!accelerated)c.payrollFreshReadSnapshot=async()=>null;return{c,calls};}
const manager={roles:['ops-manager'],current:{id:'operator'}};
async function samePreview(context=manager){const a=ctx(false),b=ctx(true),input={},options={readAcceleration:true};const before=await a.c.payrollPreviewUncached(context,input,'2026-11',true,false,options),after=await b.c.payrollPreviewUncached(context,input,'2026-11',true,false,options);assert.deepEqual(plain(after),plain(before));assert.ok(b.calls.includes('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY'));return after;}
test('one materialized source preserves exact source/cutoff versions, all-time counts and conservative month transport',async()=>{
 await setRows([lesson('oct','2026-10-31T15:59:59.999Z'),lesson('nov','2026-10-31T16:00:00.000Z'),lesson('dec','2026-11-30T16:00:00Z'),lesson('unknown',null),lesson('local','2026-10-02 12:00:00'),lesson('invalid','2026-02-30T00:00:00Z'),lesson('precision','2026-10-31T15:59:59.9999999Z')]);
 const r=await snapshot();assert.equal(r.version,(await one("SELECT xs_payroll_source_version('co') v")).v);const fullCutoff=(await one("SELECT xs_payroll_cutoff_snapshot_fast('co','2026-11') r")).r;const mapper=ctx().c.payrollSyncStatusFromCutoff;assert.deepEqual(plain(mapper(r.cutoff,true)),plain(mapper(fullCutoff,true)));assert.equal('conflicts' in r.cutoff,false);assert.equal('approval' in r.cutoff,false);assert.equal(r.total,7);assert.deepEqual(r.rows.map(x=>x.objectId),['nov','unknown','local','invalid','precision']);assert.equal(r.scopeRows.length,7);await samePreview();await samePreview({...manager,storeId:10});
});
test('cross-month and cross-store duplicate representatives keep original full ordering before JS scope/dedupe/month filtering',async()=>{
 await setRows([lesson('first','2026-10-02T00:00:00Z',{standardLessonKey:'dup',szmdid:11}),lesson('second','2026-11-02T00:00:00Z',{standardLessonKey:'dup',szmdid:10}),lesson('third',null,{szmdid:11}),lesson('fourth','2026-11-01T00:00:00Z',{storeId:'0',szmdid:11})]);
 const r=await snapshot();assert.equal(r.hasDuplicates,true);assert.deepEqual(r.rows,raw);await samePreview();const ten=await samePreview({...manager,storeId:10}),eleven=await samePreview({...manager,storeId:11});assert.equal(ten.sourceStatus.total,2);assert.equal(eleven.sourceStatus.total,2);
});
test('fresh writes and unknown source/rule fields participate in exact version; dashboard remains non-authorizing',async()=>{
 await setRows([lesson('nov','2026-11-02T00:00:00Z')]);const old=(await snapshot()).version;await query("UPDATE raw_sources SET s=s||'{\"extension\":{\"new\":true}}'::jsonb");assert.notEqual((await snapshot()).version,old);assert.equal((await snapshot()).version,(await one("SELECT xs_payroll_source_version('co') v")).v);
 const h=ctx(),result=await h.c.payrollPreviewUncached(manager,{},'2026-11',true,false,{readAcceleration:true,forDashboard:true});assert.equal(result.sourceVersion,'');assert.equal(result.sourceStatus.sync.state,'not_checked');assert.equal(result.sourceStatus.sync.settlementReady,false);const omitted=await h.c.payrollPreviewUncached(manager,{},'2026-11',true,false,{readAcceleration:true,omitSourceVersion:true});assert.equal(omitted.sourceVersion,'');
});
test('authority/parser drift returns NULL and command/default/forced-live paths never use the coherent shortcut',async()=>{
 await db.exec('BEGIN');try{await db.exec("CREATE OR REPLACE FUNCTION xs_payroll_time(v text) RETURNS timestamptz LANGUAGE sql IMMUTABLE AS $$ SELECT NULL::timestamptz $$");assert.equal(await snapshot(),null);}finally{await db.exec('ROLLBACK');}
 for(const [used,live,input,options] of [[false,false,{},{}],[true,true,{}, {readAcceleration:true}],[true,false,{targetId:'batch'},{readAcceleration:true}],[true,false,{},{}]]){const h=ctx();h.c.payrollCanonicalSnapshot=async(_c,_i,o)=>{assert.notEqual(o.readAcceleration,true);return{items:[],version:'original',sourceStatus:{sync:{}}};};h.c.Psql.one=async()=>({value:false});h.c.Psql.query=async()=>[];await h.c.payrollPreviewUncached(manager,input,'2026-11',used,live,options);assert.equal(h.calls.length,0);}
});

test('missing snapshot helper, transaction support and authority drift fall back without treating a preview as a command',async()=>{
 await setRows([lesson('present','2026-11-02T00:00:00Z')]);
 const baseline=ctx(false),expected=await baseline.c.payrollPreviewUncached(manager,{},'2026-11',true,false,{readAcceleration:true});
 for(const mode of ['missing','unsupported','null']){const h=ctx();if(mode==='unsupported')delete h.c.Psql.tx;else h.c.Psql.tx=async()=>{if(mode==='missing')throw Error('function does not exist');return null;};assert.deepEqual(plain(await h.c.payrollPreviewUncached(manager,{},'2026-11',true,false,{readAcceleration:true})),plain(expected));}
 await db.exec('BEGIN');try{await db.exec("CREATE OR REPLACE FUNCTION xs_payroll_sources(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT '{\"newAuthority\":true}'::jsonb $$");assert.equal(await snapshot(),null);}finally{await db.exec('ROLLBACK');}
});

test('conflict authority drift fails closed even when installation is repeated',async()=>{
 await db.exec('BEGIN');try{await db.exec("CREATE OR REPLACE FUNCTION xs_payroll_cutoff_conflicts(co text,mon text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT '{\"changed\":true}'::jsonb $$");assert.equal(await snapshot(),null);await db.exec(read('../sql/payroll-read-snapshot.sql'));assert.equal(await snapshot(),null);}finally{await db.exec('ROLLBACK');}
});

test('compact cutoff preserves approvals, evidence and pending-conflict sync output while excluding large current payloads',async()=>{
 await db.exec('BEGIN');try{
  await db.exec(`INSERT INTO "LegacyDirectRun" VALUES('co','verified','completed','2026-11-09T00:00:00Z');
   INSERT INTO "LegacyDirectEvidence" SELECT 'co','verified',n,'[]'::jsonb,0,md5(''),md5('') FROM unnest(ARRAY[54,59,60]) n;`);
  const check=async()=>{const compact=await snapshot(),full=(await one("SELECT xs_payroll_cutoff_snapshot_fast('co','2026-11') r")).r,mapper=ctx().c.payrollSyncStatusFromCutoff;assert.deepEqual(plain(mapper(compact.cutoff,true)),plain(mapper(full,true)));assert.equal(compact.cutoff.version,full.version);assert.equal('conflicts' in compact.cutoff,false);assert.equal('approval' in compact.cutoff,false);return compact.cutoff;};
  let value=await check();assert.equal(value.complete,true);await query('INSERT INTO "PayrollCutoffApproval" VALUES($1,$2,$3)',['co','2026-11',value.version]);value=await check();assert.equal(value.approved,true);
  await query('INSERT INTO test_conflicts VALUES($1::jsonb)',[JSON.stringify({id:'issue',resolved:false,current:{large:'x'.repeat(100000)}})]);value=await check();assert.equal(value.pending,1);assert.equal(value.approved,false);
 }finally{await db.exec('ROLLBACK');}
});

test('thin user hydration preserves phone-only and legacy string aliases',async()=>{
 await db.exec('BEGIN');try{
  await db.exec(`ALTER TABLE "_User" ADD COLUMN phone text;ALTER TABLE "_User" ADD COLUMN nickname2 text;ALTER TABLE "_User" ADD COLUMN legacyuserdata jsonb;
  INSERT INTO "_User"("objectId",company,"legacyUserId","legacyUserData",phone,nickname2) VALUES('phone-only','co',91,'"{\\"ParentUserID\\":10,\\"UserID\\":91}"','13800000091','昵称二');`);
  const c=ctx().c,full=await query('SELECT * FROM "_User" WHERE company=$1',['co']),thin=await c.payrollReadUsers('co');
  const summary=rows=>rows.map(row=>[row.objectId,c.displayName(row),c.mobileOf(row),c.legacyOf(row)]).sort((a,b)=>a[0].localeCompare(b[0]));
  assert.deepEqual(plain(summary(thin)),plain(summary(full)));
 }finally{await db.exec('ROLLBACK');}
});