xiaoshu-app/scripts/tests/payroll-source-bulk.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 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_bulk',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_bulk($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'));
});
after(()=>db.close());

test('bulk projection preserves first valid appointment, tombstones, zero snapshots, text keys, tenants and exact source hash',async()=>{
 assert.equal((await q("SELECT to_regprocedure('xs_payroll_sources_bulk(text)') IS NOT NULL AS available"))[0].available,true,'bulk helper is installed');
 await equal();
 const result=await rows('xs_payroll_sources_bulk');
 assert.equal(result.find(r=>r.objectId==='l').teacherPaySnapshot,0);
 assert.equal(result.find(r=>r.objectId==='l').commonStatus,-2);
 assert.equal(result.find(r=>r.objectId==='l').legacyGeneralId,200);
 assert.equal(result.find(r=>r.objectId==='l-plain').teacherPaySnapshot,null);
 assert.equal(result.some(r=>r.objectId==='suppressed'),false);
 assert.equal(result.find(r=>r.objectId==='fact').teacherPaySnapshot,40);
});

test('bulk projection reads changes without any result cache and includes schema extensions',async()=>{
 await db.exec(`BEGIN;UPDATE "CourseAppointment" SET "teacherPaySnapshot"=77 WHERE company='co';ALTER TABLE "LessonRecord" ADD COLUMN extension jsonb DEFAULT '{"new":true}';ALTER TABLE "LessonRecordFact" ADD COLUMN extension text DEFAULT 'new';`);
 try{await equal();assert.equal((await rows('xs_payroll_sources_bulk')).find(r=>r.objectId==='fact').teacherPaySnapshot,77);assert.deepEqual((await rows('xs_payroll_sources_bulk')).find(r=>r.objectId==='l').extension,{new:true});}finally{await db.exec('ROLLBACK');}
 await equal();
});

test('changed authority falls back to the current authoritative definition',async()=>{
 await db.exec(`BEGIN;CREATE OR REPLACE FUNCTION xs_payroll_sources(co text) RETURNS SETOF jsonb LANGUAGE sql AS $$ SELECT jsonb_build_object('newAuthority',co) $$;`);
 try{assert.deepEqual(await rows('xs_payroll_sources_bulk'),[{newAuthority:'co'}]);}finally{await db.exec('ROLLBACK');}
});

test('ambiguous lesson identities and appointment IDs preserve the authoritative representative order',async()=>{
 for(const mutation of [
  `INSERT INTO "LessonRecord" SELECT (jsonb_populate_record(NULL::"LessonRecord",to_jsonb(l)||'{"objectId":"duplicate","id":10,"legacyGeneralId":200}'::jsonb)).* FROM "LessonRecord" l WHERE "objectId"='l'`,
  `INSERT INTO "CourseAppointment" SELECT (jsonb_populate_record(NULL::"CourseAppointment",to_jsonb(a)||'{"objectId":"duplicate","teacherPaySnapshot":80}'::jsonb)).* FROM "CourseAppointment" a WHERE "objectId"='a'`,
  `INSERT INTO "LessonRecordFact" SELECT (jsonb_populate_record(NULL::"LessonRecordFact",to_jsonb(f)||'{"objectId":"duplicate"}'::jsonb)).* FROM "LessonRecordFact" f WHERE "objectId"='fact'`,
  // JSON renders this large ID in decimal, while float::text uses an exponent.
  // The original textual anti-join retains both rows, making their standard keys ambiguous.
  `INSERT INTO "LessonRecord"("objectId",company,"legacyGeneralId") VALUES('huge-lesson','co',1e30);
   INSERT INTO "LessonRecordFact"("objectId",company,"legacyGeneralId") VALUES('huge-fact','co',1e30)`
 ]){
  await db.exec('BEGIN');
  try{
   await db.exec(mutation);
   // A different authoritative order makes bypassing the ambiguity fallback observable.
   await db.exec(`ALTER FUNCTION xs_payroll_sources(text) RENAME TO xs_payroll_sources_reference;
   CREATE FUNCTION xs_payroll_sources(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$ SELECT s FROM xs_payroll_sources_reference(co) s ORDER BY s->>'objectId' DESC $$;
   UPDATE "PayrollReadAuthority" SET hash=md5(pg_get_functiondef('xs_payroll_sources(text)'::regprocedure)) WHERE name='xs_payroll_sources(text)';`);
   await equal();
  }finally{await db.exec('ROLLBACK');}
 }
});