xiaoshu-app/scripts/tests/payroll-source-index.test.mjs

import {test,before,after} from 'node:test';
import assert from 'node:assert/strict';
import {readFileSync,existsSync} from 'node:fs';
import vm from 'node:vm';
import {PGlite} from '@electric-sql/pglite';

const db=new PGlite();
const q=async(sql,args=[])=>(await db.query(sql,args)).rows;
const source=readFileSync(new URL('../deploy-admin-functions.mjs',import.meta.url),'utf8');
const integrity=readFileSync(new URL('../sql/payroll-integrity.sql',import.meta.url),'utf8');
const incremental=readFileSync(new URL('../sql/payroll-incremental-reads.sql',import.meta.url),'utf8');
const indexPath=new URL('../sql/payroll-source-index.sql',import.meta.url);
const bulkPath=new URL('../sql/payroll-source-bulk.sql',import.meta.url);
const sqlFunction=(text,name)=>text.match(new RegExp('CREATE OR REPLACE FUNCTION '+name.replaceAll('(','\\(').replaceAll(')','\\)')+'[\\s\\S]*?\\$\\$;'))[0];
const rows=async(name,co='co')=>(await q(`SELECT s FROM ${name}($1) s`,[co])).map(r=>r.s);
async function equal(){for(const co of ['co','other','empty']){
 const a=await rows('xs_payroll_sources',co),b=await rows('xs_payroll_sources_indexed',co);
 assert.deepEqual(b,a);
 const hashes=await q(`SELECT (SELECT md5(string_agg(s::text,'|' ORDER BY s::text)) FROM xs_payroll_sources($1) s) original,(SELECT md5(string_agg(s::text,'|' ORDER BY s::text)) FROM xs_payroll_sources_indexed($1) s) bulk`,[co]);
 assert.equal(hashes[0].bulk,hashes[0].original);
}}
before(async()=>{
 const block=source.slice(source.indexOf('async function ensurePayrollSchemas()'));
 const schemas=vm.runInNewContext('('+block.match(/const schemas = (\{[\s\S]*?\n  });/)[1]+')');
 for(const name of ['LessonRecord','LessonRecordFact','CourseAppointment']){
  const cols={objectId:'text PRIMARY KEY',createdAt:'timestamptz',updatedAt:'timestamptz',...Object.fromEntries(Object.entries(schemas[name]).map(([k,v])=>[k,({Number:'double precision',Date:'timestamptz',Boolean:'boolean',Array:'jsonb',Object:'jsonb'})[v.type]||'text']))};
  await db.exec(`CREATE TABLE "${name}" (${Object.entries(cols).map(([k,v])=>`"${k}" ${v}`).join(',')})`);
 }
 await db.exec(`CREATE TABLE "CommonModel" ("objectId" text PRIMARY KEY,company text,"generalId" numeric,"itemId" numeric,"modelId" numeric,"createdAt" timestamptz,status numeric,title text,subtitle text);
 ALTER TABLE "CourseAppointment" ADD COLUMN jffs text,ADD COLUMN jssj text;
 ALTER TABLE "LessonRecord" ADD COLUMN "courseCategoryKey" text,ADD COLUMN "courseCategoryName" text,ADD COLUMN "teacherPaySnapshot" double precision;
 CREATE TABLE "PayrollReadAuthority"(name text PRIMARY KEY,hash text);
 CREATE FUNCTION xs_payroll_source_version(co text) RETURNS text LANGUAGE sql AS $$ SELECT co $$;
 CREATE FUNCTION xs_payroll_cutoff_snapshot(co text,mon text) RETURNS jsonb LANGUAGE sql AS $$ SELECT '{}'::jsonb $$;`);
 await db.exec(sqlFunction(integrity,'xs_payroll_time'));
 await db.exec(sqlFunction(integrity,'xs_payroll_sources'));
 await db.exec(sqlFunction(incremental,'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 "CourseAppointment"("objectId",company,id,"teacherPaySnapshot","courseCategoryKey",jssj) VALUES ('a','co',1,40,'word','2026-10-01 01:00:00'),('other-a','other',1,99,'other','2026-10-02');
 INSERT INTO "CommonModel"("objectId",company,"modelId","itemId","generalId",status,title) VALUES
 ('a-missing','co',54,999,100,99,''),('b-appt','co',54,1,100,99,''),('other-appt','other',54,1,100,99,''),
 ('a-lesson','co',59,1,200,-2,'saved student'),('z-lesson','co',59,1,999,99,'wrong student');
 INSERT INTO "LessonRecord"("objectId",company,id,"legacyGeneralId",yyds,"teacherPaySnapshot",plpjsj) VALUES
 ('l','co',1,201,'100',0,'invalid'),('l-plain','co',2,0,'0100',null,'2026-09-30 23:59:59'),('other-l','other',1,300,'100',null,'2026-10-02');
 INSERT INTO "LessonRecordFact"("objectId",company,"legacyGeneralId",yyds,plpjsj) VALUES ('suppressed','co',200,'100','2026-09-01'),('fact','co',400,'100','2026-10-03'),('fact-zero','co',0,'100','2026-10-04');`);
 if(existsSync(bulkPath))await db.exec(readFileSync(bulkPath,'utf8'));
 await db.exec(`CREATE TABLE "MemoryPracticeRecord"("objectId" text PRIMARY KEY,company text,id numeric);
 CREATE TABLE "TeachingCompensationRule"("objectId" text PRIMARY KEY,company text,amount numeric);
 CREATE TABLE "CourseTeachingProfile"("objectId" text PRIMARY KEY,company text,category text);
 CREATE TABLE "CoachPayRate"("objectId" text PRIMARY KEY,company text,amount numeric);
 CREATE TABLE "ReviewRewardIncome"("objectId" text PRIMARY KEY,company text,status text,amount numeric);
 SET TimeZone='Etc/GMT-8';SET extra_float_digits=1;`);
 await db.exec(sqlFunction(integrity,'xs_payroll_source_version'));
 await db.exec(`UPDATE "PayrollReadAuthority" SET hash=md5(pg_get_functiondef(to_regprocedure(name)));`);
 await db.exec(incremental);
 if(existsSync(indexPath))await db.exec(readFileSync(indexPath,'utf8'));
});
after(()=>db.close());

test('validated projection index returns exact authoritative JSON and source versions',async()=>{
 assert.equal((await q("SELECT to_regprocedure('xs_payroll_source_entries(text)') IS NOT NULL AS available"))[0].available,true,'source entries helper installed');
 await equal();
 const [v]=await q("SELECT xs_payroll_source_version('co') original,xs_payroll_source_version_indexed('co') indexed");assert.equal(v.indexed,v.original);
 const entries=await q("SELECT * FROM xs_payroll_source_entries('co')");
 assert.equal(entries.length,4);assert.deepEqual(entries.map(x=>x.ordinal),[1,2,3,4]);
 for(const entry of entries)assert.deepEqual(JSON.parse(entry.row_text),entry.row_data);
});

test('current roots and dependency signatures reject missing, orphaned and stale projection entries',async()=>{
 await db.exec('BEGIN');
 try{
  const saved=(await q(`SELECT * FROM "PayrollSourceIndex" WHERE company='co' AND kind='fact' AND "objectId"='fact'`))[0];
  await db.exec(`UPDATE "CourseAppointment" SET "teacherPaySnapshot"=77 WHERE company='co';`);
  // Simulate an older concurrent transaction winning the best-effort projection write.
  await q(`UPDATE "PayrollSourceIndex" SET row_data=$1::jsonb,row_text=$2,input_signature=$3 WHERE company='co' AND kind='fact' AND "objectId"='fact'`,[JSON.stringify(saved.row_data),saved.row_text,saved.input_signature]);
  await db.exec(`DELETE FROM "PayrollSourceIndex" WHERE "objectId"='l-plain';
   INSERT INTO "PayrollSourceIndex"(kind,company,"objectId",row_data,row_text,input_signature,generation) SELECT kind,company,'orphan',row_data,row_text,input_signature,generation FROM "PayrollSourceIndex" WHERE "objectId"='fact';`);
  assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,true);
  await equal();assert.equal((await rows('xs_payroll_sources_indexed')).find(r=>r.objectId==='fact').teacherPaySnapshot,77);
 }finally{await db.exec('ROLLBACK');}
});

test('incremental dependencies include canonical suppression, mapping reselection, tenant moves and rollback',async()=>{
 const mutations=[
  `UPDATE "CommonModel" SET "generalId"=400 WHERE "objectId"='a-lesson'`,
  `DELETE FROM "CommonModel" WHERE "objectId"='a-lesson'`,
  `UPDATE "CommonModel" SET "itemId"=999 WHERE "objectId"='b-appt'`,
  `UPDATE "CourseAppointment" SET company='other',id=9 WHERE "objectId"='a'`,
  `UPDATE "LessonRecord" SET company='other' WHERE "objectId"='l'`,
  `DELETE FROM "LessonRecord" WHERE "objectId"='l'`,
  `UPDATE "LessonRecordFact" SET company='other',"legacyGeneralId"=300 WHERE "objectId"='fact'`
 ];
 for(const sql of mutations){await db.exec('BEGIN');try{await db.exec(sql);await equal();}finally{await db.exec('ROLLBACK');}await equal();}
});

test('best-effort upkeep updates only affected targets with signatures matching native rows',async()=>{
 await db.exec('BEGIN');try{
  const untouched=(await q(`SELECT ctid::text AS address FROM "PayrollSourceIndex" WHERE "objectId"='l-plain'`))[0].address;
  await db.exec(`UPDATE "CourseAppointment" SET "teacherPaySnapshot"=77 WHERE "objectId"='a';`);
  assert.equal((await q(`SELECT row_data->>'teacherPaySnapshot' AS amount FROM "PayrollSourceIndex" WHERE "objectId"='fact'`))[0].amount,'77');
  assert.equal((await q(`SELECT ctid::text AS address FROM "PayrollSourceIndex" WHERE "objectId"='l-plain'`))[0].address,untouched);
  const [mismatch]=await q(`SELECT count(*)::int AS n FROM "PayrollSourceIndex" i CROSS JOIN LATERAL xs_payroll_source_index_target(i.company,i.kind,i."objectId") current WHERE i.input_signature IS DISTINCT FROM current.input_signature`);
  assert.equal(mismatch.n,0);await equal();
 }finally{await db.exec('ROLLBACK');}
});

test('schema and trigger drift fail closed; truncate requires a new generation and full reseed',async()=>{
 await db.exec('BEGIN');try{
  await db.exec(`ALTER TABLE "LessonRecord" ADD COLUMN extension text DEFAULT 'new';`);
  assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,false);await equal();
 }finally{await db.exec('ROLLBACK');}
 await db.exec('BEGIN');try{
  await db.exec(`ALTER TABLE "CourseAppointment" DISABLE TRIGGER xs_payroll_source_index_row;UPDATE "CourseAppointment" SET "teacherPaySnapshot"=77;ALTER TABLE "CourseAppointment" ENABLE ALWAYS TRIGGER xs_payroll_source_index_row;`);
  assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,false);await equal();
 }finally{await db.exec('ROLLBACK');}
 await db.exec('BEGIN');try{
  const old=(await q('SELECT generation FROM "PayrollSourceIndexMeta"'))[0].generation;
  await db.exec('TRUNCATE "LessonRecord"');assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,false);await equal();
  await db.exec('SAVEPOINT invalid_reenable');
  await assert.rejects(db.exec('UPDATE "PayrollSourceIndexMeta" SET valid=true'),/reseed/);
  await db.exec('ROLLBACK TO SAVEPOINT invalid_reenable');
  await db.exec('SELECT xs_payroll_source_index_rebuild()');assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,true);
  assert.notEqual((await q('SELECT generation FROM "PayrollSourceIndexMeta"'))[0].generation,old);await equal();
 }finally{await db.exec('ROLLBACK');}
});

test('generation mismatches miss safely and changed financial authority bypasses the index',async()=>{
 await db.exec('BEGIN');try{
  await db.exec(`UPDATE "PayrollSourceIndex" SET generation='old-generation';`);await equal();
  await db.exec(`CREATE OR REPLACE FUNCTION xs_payroll_sources(co text) RETURNS SETOF jsonb LANGUAGE sql AS $$ SELECT jsonb_build_object('newAuthority',co) $$;
   CREATE OR REPLACE FUNCTION xs_payroll_source_version(co text) RETURNS text LANGUAGE sql AS $$ SELECT 'new-policy:'||co $$;`);
  assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,false);
  assert.deepEqual(await rows('xs_payroll_sources_indexed'),[{newAuthority:'co'}]);
  assert.equal((await q("SELECT xs_payroll_source_version_indexed('co') AS v"))[0].v,'new-policy:co');
 }finally{await db.exec('ROLLBACK');}
});

test('version still recomputes all rules and earned reward text exactly',async()=>{
 await db.exec('BEGIN');try{
  await db.exec(`INSERT INTO "TeachingCompensationRule" VALUES('r','co',40);
   INSERT INTO "CourseTeachingProfile" VALUES('p','co','word');INSERT INTO "CoachPayRate" VALUES('r','co',0);
   INSERT INTO "ReviewRewardIncome" VALUES('earned','co','earned',0.50),('pending','co','pending',99);`);
  const initial=(await q("SELECT xs_payroll_source_version_indexed('co') AS v"))[0].v;
  for(const mutation of [`UPDATE "TeachingCompensationRule" SET amount=41`,`UPDATE "CourseTeachingProfile" SET category='changed'`,`UPDATE "CoachPayRate" SET amount=1`,`UPDATE "ReviewRewardIncome" SET amount=1 WHERE status='earned'`]){
   await db.exec(mutation);const [v]=await q("SELECT xs_payroll_source_version('co') original,xs_payroll_source_version_indexed('co') indexed");assert.equal(v.indexed,v.original);assert.notEqual(v.indexed,initial);
  }
 }finally{await db.exec('ROLLBACK');}
});

test('version-only entries omit JSON while retaining exact text on hits and misses',async()=>{
 assert.equal((await q("SELECT to_regprocedure('xs_payroll_source_index_entries(text,boolean,boolean)') IS NOT NULL AS available"))[0].available,true);
 await db.exec('BEGIN');try{
  for(const clear of [false,true]){
   if(clear)await db.exec('DELETE FROM "PayrollSourceIndex"');
   const entries=await q("SELECT * FROM xs_payroll_source_index_entries('co',true,false)");
   assert(entries.every(e=>e.row_data===null));
   assert.deepEqual(entries.map(e=>JSON.parse(e.row_text)),await rows('xs_payroll_sources'));
   const [v]=await q("SELECT xs_payroll_source_version('co') original,xs_payroll_source_version_indexed('co') indexed");assert.equal(v.indexed,v.original);
  }
 }finally{await db.exec('ROLLBACK');}
});

test('unrelated schemas with matching trigger names do not invalidate the production generation',async()=>{
 await db.exec('BEGIN');try{
  await db.exec(`CREATE SCHEMA diagnostic;CREATE TABLE diagnostic.copy("objectId" text);
   CREATE TRIGGER xs_payroll_source_index_row AFTER INSERT ON diagnostic.copy FOR EACH ROW EXECUTE FUNCTION xs_payroll_source_index_row();`);
  assert.equal((await q('SELECT xs_payroll_source_index_ready() ready'))[0].ready,true);
 }finally{await db.exec('ROLLBACK');}
});

test('scoped entries keep all version text and scope metadata while hydrating only month or coach candidates',async()=>{
 assert.equal((await q("SELECT to_regprocedure('xs_payroll_source_entries_scoped(text,text,text,text[],boolean)') IS NOT NULL AS available"))[0].available,true);
 await db.exec('BEGIN');try{
  await db.exec(`UPDATE "LessonRecord" SET jsmz=CASE WHEN "objectId"='l' THEN '30' ELSE '31' END WHERE company='co';UPDATE "LessonRecordFact" SET jsmz='30' WHERE company='co';`);
  const original=await rows('xs_payroll_sources');
  for(const clear of [false,true]){
   if(clear)await db.exec('DELETE FROM "PayrollSourceIndex"');
   const selected=await q("SELECT * FROM xs_payroll_source_entries_scoped('co','2026-10',NULL,NULL,true) ORDER BY ordinal");
   assert.equal(selected.length,original.length);
   assert.deepEqual(selected.map(r=>JSON.parse(r.row_text)),original);
   assert.deepEqual(selected.map(r=>r.scope_data),original.map(r=>[r.coachId??null,r.jsmz??null,r.storeId??null,r.szmdid??null]));
   assert.equal(selected.find(r=>r.standard_key==='lesson:l-plain').row_data,null);
   assert.equal(selected.find(r=>r.standard_key==='legacy:200').row_data.objectId,'l');
   assert.equal(selected.find(r=>r.standard_key==='legacy:400').month_candidate_key,null,'unknown dates remain candidates');
   const teacher=await q("SELECT * FROM xs_payroll_source_entries_scoped('co',NULL,'30',ARRAY['100'],false) ORDER BY ordinal");
   assert(teacher.every(r=>r.row_text===null));
   assert.deepEqual(teacher.filter(r=>r.row_data!==null).map(r=>r.row_data),original.filter(r=>(r.coachId??r.jsmz)==='30'&&(r.appointmentGeneralId??r.yyds)==='100'));
  }
 }finally{await db.exec('ROLLBACK');}
});

test('hidden fact projections are maintained independently and root membership alone suppresses or restores them',async()=>{
 await db.exec('BEGIN');try{
  await db.exec("SELECT xs_payroll_source_index_put('co','fact','suppressed')");
  assert.equal((await q(`SELECT count(*)::int n FROM "PayrollSourceIndex" WHERE "objectId"='suppressed'`))[0].n,1);
  assert.equal((await rows('xs_payroll_sources_indexed')).some(r=>r.objectId==='suppressed'),false);
  const hidden=(await q(`SELECT ctid::text address FROM "PayrollSourceIndex" WHERE "objectId"='suppressed'`))[0].address;
  await db.exec(`DELETE FROM "LessonRecord" WHERE "objectId"='l'`);
  assert.equal((await q(`SELECT ctid::text address FROM "PayrollSourceIndex" WHERE "objectId"='suppressed'`))[0].address,hidden,'lesson membership changes do not rewrite fact projections');
  await equal();assert.equal((await rows('xs_payroll_sources_indexed')).some(r=>r.objectId==='suppressed'),true);
 }finally{await db.exec('ROLLBACK');}
});