xiaoshu-app/scripts/tests/payroll-cutoff-bulk.test.mjs

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

const read = path => fs.readFileSync(new URL(path, import.meta.url), 'utf8');
const originalSql = read('../sql/payroll-cutoff.sql');
const extract = (source, name) => {
  const match = source.match(new RegExp('CREATE OR REPLACE FUNCTION ' + name + '\\([\\s\\S]*?\\$\\$;'));
  assert.ok(match, `Missing actual SQL function ${name}`);
  return match[0];
};
const oracleSql = extract(originalSql, 'xs_payroll_cutoff_conflicts');
const authorityInstall = read('../sql/payroll-read-snapshot.sql').match(/DO \$install\$[\s\S]*?\$install\$;/)[0];
const db = new PGlite();
const query = async (sql, params = []) => (await db.query(sql, params)).rows;
const one = async (sql, params = []) => (await query(sql, params))[0];
const tables = { 54: 'CourseAppointment', 59: 'LessonRecord', 60: 'MemoryPracticeRecord' };
const month = '2026-11';
let nextId = 1;

before(async () => {
  await db.exec(`
    CREATE TABLE "LegacySyncConflict" ("objectId" text PRIMARY KEY, company text, dataset text, "legacyKey" text, resolved boolean, message text);
    CREATE TABLE "CommonModel" ("objectId" text, company text, "generalId" numeric, "modelId" int, "itemId" numeric, title text, subtitle text);
    CREATE TABLE "CourseAppointment" (company text, id numeric, yysj text, "teacherPaySnapshot" numeric, extension jsonb);
    CREATE TABLE "LessonRecord" (company text, id numeric, "lessonAt" text, plpjsj text, "teacherPaySnapshot" numeric, extension jsonb);
    CREATE TABLE "MemoryPracticeRecord" (company text, id numeric, wcsj text, kywrq text, "teacherPaySnapshot" numeric, extension jsonb);
    CREATE TABLE "LegacyDirectRun" (company text, "runId" text, status text, "finishedAt" timestamptz);
    CREATE TABLE "PayrollCutoffDecision" (company text, month text, "conflictId" text, fingerprint text, PRIMARY KEY(company,month,"conflictId"));
    CREATE FUNCTION xs_payroll_authority_ready() RETURNS boolean LANGUAGE sql STABLE AS $$ SELECT true $$;
    SET timezone='UTC';
  `);
  await db.exec(extract(read('../sql/payroll-integrity.sql'), 'xs_payroll_time'));
  await db.exec(oracleSql);
  await db.exec(authorityInstall);
  await db.exec(read('../sql/payroll-cutoff-bulk.sql'));
});
beforeEach(async () => {
  nextId = 1;
  await db.exec('BEGIN');
  await db.exec(`INSERT INTO "LegacyDirectRun" VALUES
    ('co','latest','completed-with-preserved-conflicts','2026-11-09T00:00:00Z'),
    ('co','old','completed','2026-11-01T00:00:00Z'),
    ('co','not-complete','running','2026-11-10T00:00:00Z'),
    ('other','other-latest','completed','2026-12-01T00:00:00Z');`);
});
afterEach(() => db.exec('ROLLBACK'));
after(() => db.close());

async function add({ id, dataset = 59, company = 'co', resolved = false, message = '{"fields":["teacherPaySnapshot"],"reason":"金额待核验"}', date = '2026-11-01 12:00:00', fallback = null, amount = 40.5, missingMapping = false, missingAddon = false, title = '学员', subtitle = '课程' } = {}) {
  const key = nextId++, conflictId = id ?? `conflict-${String(key).padStart(3, '0')}`;
  await query('INSERT INTO "LegacySyncConflict" VALUES ($1,$2,$3,$4,$5,$6)', [conflictId, company, String(dataset), String(key), resolved, message]);
  if (!missingMapping) await query('INSERT INTO "CommonModel" VALUES ($1,$2,$3,$4,$5,$6,$7)', ['cm-' + key, company, key, dataset, key, title, subtitle]);
  if (!missingAddon && !missingMapping && tables[dataset]) {
    const columns = dataset === 54 ? 'yysj' : dataset === 59 ? '"lessonAt",plpjsj' : 'wcsj,kywrq';
    const dates = dataset === 54 ? [date] : [date, fallback];
    const values = [company, key, ...dates, amount, JSON.stringify({ source: 'native', arbitrary: [null, 0, '保留'] })];
    await query(`INSERT INTO "${tables[dataset]}" (company,id,${columns},"teacherPaySnapshot",extension) VALUES (${values.map((_, i) => '$' + (i + 1)).join(',')})`, values);
  }
  return { id: conflictId, key, dataset };
}

const serialized = async (fn, co = 'co', mon = month) => (await one(`SELECT COALESCE(jsonb_agg(value ORDER BY ord),'[]'::jsonb)::text AS value FROM ${fn}($1,$2) WITH ORDINALITY AS c(value,ord)`, [co, mon])).value;
const rows = async (fn, co = 'co', mon = month) => JSON.parse(await serialized(fn, co, mon));
async function same(co = 'co', mon = month) {
  const beforeText = await serialized('xs_payroll_cutoff_conflicts', co, mon);
  const afterText = await serialized('xs_payroll_cutoff_conflicts_bulk', co, mon);
  assert.equal(afterText, beforeText, 'Exact ordered PostgreSQL JSON must match');
  const before = JSON.parse(beforeText), after = JSON.parse(afterText);
  // Includes current payment amounts, every unknown column, fields order, output
  // order, fingerprint, cosmetic classification and the financial blocking state.
  assert.deepEqual(after, before);
  assert.deepEqual(after.map(row => row.fingerprint), before.map(row => row.fingerprint));
  return after;
}

// Observe the fallback branch without replacing the oracle's semantics. The
// accepted authority delegates to the actual original PL/pgSQL function.
async function trackLegacyCalls() {
  await db.exec(oracleSql.replace('FUNCTION xs_payroll_cutoff_conflicts(', 'FUNCTION xs_payroll_cutoff_conflicts_oracle('));
  await db.exec(`CREATE OR REPLACE FUNCTION xs_payroll_cutoff_conflicts(co text,mon text) RETURNS SETOF jsonb LANGUAGE plpgsql AS $$
    BEGIN
      PERFORM set_config('test.legacy_calls',(COALESCE(NULLIF(current_setting('test.legacy_calls',true),''),'0')::int+1)::text,true);
      RETURN QUERY SELECT s FROM xs_payroll_cutoff_conflicts_oracle(co,mon) s;
    END $$;
    DROP FUNCTION xs_payroll_read_snapshot_authority_ready();`);
  await db.exec(authorityInstall);
  assert.equal((await one('SELECT xs_payroll_read_snapshot_authority_ready() ready')).ready, true);
}
async function compareFallback(expectedCalls) {
  const expected = await rows('xs_payroll_cutoff_conflicts');
  await query("SELECT set_config('test.legacy_calls','0',true)");
  assert.deepEqual(await rows('xs_payroll_cutoff_conflicts_bulk'), expected);
  assert.equal(Number((await one("SELECT current_setting('test.legacy_calls') n")).n), expectedCalls);
}

test('actual original and bulk preserve ordered conflicts, all three payment payloads, missing addons and tenant/status filters', async () => {
  await add({ id: 'z-review', dataset: 60, amount: 1.25 });
  await add({ id: 'a-appointment', dataset: 54, amount: 0 });
  await add({ id: 'm-lesson', dataset: 59, amount: 40.5 });
  await add({ id: 'b-missing-addon', missingAddon: true });
  await add({ id: 'c-missing-mapping', missingMapping: true });
  await add({ id: 'ignored-resolved', resolved: true });
  await add({ id: 'ignored-dataset', dataset: 52 });
  await add({ id: 'ignored-tenant', company: 'other' });
  const result = await same();
  assert.deepEqual(result.map(row => row.id), ['a-appointment', 'b-missing-addon', 'c-missing-mapping', 'm-lesson', 'z-review']);
  assert.equal(result[0].current.teacherPaySnapshot, 0);
  assert.equal(result[3].current.teacherPaySnapshot, 40.5);
  assert.equal(result[4].current.teacherPaySnapshot, 1.25);
  assert.equal(result[1].current, null); assert.equal(result[2].current, null);
  assert.equal(result.filter(row => !row.resolved).length, 5);
  assert.ok(result.every(row => /^[a-f0-9]{32}$/.test(row.fingerprint)));
  await trackLegacyCalls(); await compareFallback(0); // Prove the ordinary case uses bulk SQL.
});

test('malformed, null and scalar messages/fields retain authoritative cosmetic and pending decisions', async () => {
  const cases = [
    [null, false], ['not json', false], ['null', false], ['42', false], ['"text"', false], ['[]', false],
    ['{}', false], ['{"fields":null}', false], ['{"fields":true}', false], ['{"fields":"title"}', false],
    ['{"fields":{}}', false], ['{"fields":[]}', false], ['{"fields":["title"]}', true],
    ['{"fields":["inputer","title","inputer"],"reason":17}', true], ['{"fields":[null]}', true],
    ['{"fields":["title",null]}', true], ['{"fields":["title","teacherPaySnapshot"]}', false],
    ['{"fields":[{},2,true]}', false],
  ];
  for (const [index, [message]] of cases.entries()) await add({ id: String(index).padStart(2, '0'), message });
  const result = await same();
  assert.deepEqual(result.map(row => row.cosmetic), cases.map(([, cosmetic]) => cosmetic));
  assert.deepEqual(result.map(row => row.resolved), cases.map(([, cosmetic]) => cosmetic));
  assert.equal(result[13].reason, '17');
});

test('future months alone are excluded, with the real Shanghai parser, fallback dates and malformed dates', async () => {
  const fixtures = [
    { id: 'old', date: '2026-01-01' },
    { id: 'local-end', date: '2026-11-30 23:59:59' },
    { id: 'utc-before', date: '2026-11-30T15:59:59.999Z' },
    { id: 'utc-next', date: '2026-11-30T16:00:00Z' },
    { id: 'local-next', date: '2026-12-01 00:00:00' },
    { id: 'invalid-future', date: '2099-02-30' },
    { id: 'invalid', date: 'not-a-date' },
    { id: 'unknown', date: null },
    { id: 'empty', date: '' },
    { id: 'whitespace', date: '   ' },
  ];
  for (const dataset of [54, 59, 60]) for (const item of fixtures) await add({ ...item, id: `${dataset}-${item.id}`, dataset });
  await add({ id: 'lesson-fallback-old', date: '', fallback: '2026-10-01' });
  await add({ id: 'lesson-fallback-future', date: null, fallback: '2026-12-01' });
  await add({ id: 'review-fallback-old', dataset: 60, date: '', fallback: '2026-10-01' });
  await add({ id: 'review-fallback-future', dataset: 60, date: null, fallback: '2026-12-01' });
  // Nonempty but invalid primary dates must not silently use a valid fallback.
  await add({ id: 'invalid-primary-kept', date: 'invalid', fallback: '2026-12-01' });
  const result = await same();
  assert.equal(result.length, 27);
  assert.ok(result.every(row => !row.id.includes('utc-next') && !row.id.includes('local-next') && !row.id.includes('fallback-future')));
  assert.ok(result.some(row => row.id === 'invalid-primary-kept'));
  assert.equal(result.find(row => row.id === 'lesson-fallback-old').date, '2026-10-01');
});

test('exact conflict decisions resolve only their company, month and current fingerprint; fresh changes revoke them', async () => {
  const item = await add({ id: 'financial-conflict', dataset: 59, amount: 40.5 });
  const initial = (await same())[0];
  await query('INSERT INTO "PayrollCutoffDecision" VALUES ($1,$2,$3,$4)', ['other', month, item.id, initial.fingerprint]);
  await query('INSERT INTO "PayrollCutoffDecision" VALUES ($1,$2,$3,$4)', ['co', '2026-10', item.id, initial.fingerprint]);
  assert.equal((await same())[0].resolved, false);
  await query('INSERT INTO "PayrollCutoffDecision" VALUES ($1,$2,$3,$4)', ['co', month, item.id, initial.fingerprint]);
  assert.equal((await same())[0].resolved, true);
  const edits = [
    ['UPDATE "LessonRecord" SET "teacherPaySnapshot"=41.75 WHERE id=$1', [item.key]],
    ['UPDATE "CommonModel" SET title=$1 WHERE "itemId"=$2', ['姓名改正', item.key]],
    ['UPDATE "CommonModel" SET subtitle=$1 WHERE "itemId"=$2', ['课程改正', item.key]],
    ['UPDATE "LegacySyncConflict" SET message=$1 WHERE "objectId"=$2', ['{"fields":["amount"],"reason":"新差异"}', item.id]],
    ['INSERT INTO "LegacyDirectRun" VALUES($1,$2,$3,$4)', ['co', 'new-complete', 'completed', '2026-11-11T00:00:00Z']],
  ];
  let prior = initial.fingerprint;
  for (const [sql, params] of edits) {
    await query(sql, params);
    const changed = (await same())[0];
    assert.notEqual(changed.fingerprint, prior); assert.equal(changed.resolved, false);
    await query('UPDATE "PayrollCutoffDecision" SET fingerprint=$1 WHERE company=$2 AND month=$3 AND "conflictId"=$4', [changed.fingerprint, 'co', month, item.id]);
    assert.equal((await same())[0].resolved, true);
    prior = changed.fingerprint;
  }
  assert.equal((await same())[0].current.teacherPaySnapshot, 41.75);
});

for (const dataset of [54, 59, 60]) test(`duplicate addon id in dataset ${dataset} delegates to original SELECT INTO semantics`, async () => {
  await add({ id: 'subject', dataset });
  await db.exec(`INSERT INTO "${tables[dataset]}" SELECT * FROM "${tables[dataset]}"`);
  // Make the duplicate materially different: a bulk join must neither multiply
  // payable amounts nor silently pick a different row from the original.
  await db.exec(`UPDATE "${tables[dataset]}" SET "teacherPaySnapshot"=999.75,extension='{"duplicate":"different payment"}' WHERE ctid=(SELECT max(ctid) FROM "${tables[dataset]}")`);
  await trackLegacyCalls();
  await compareFallback(1);
  assert.equal((await same()).length, 1);
});

test('duplicate addon ids are scoped to the tenant and null ids cannot match', async () => {
  await add({ id: 'subject' });
  await db.exec(`INSERT INTO "LessonRecord"(company,id) VALUES('other',5),('other',5),('co',NULL),('co',NULL)`);
  await trackLegacyCalls(); await compareFallback(0);
});

test('missing completed runs and null resolved flags keep unresolved financial rows', async () => {
  await db.exec('DELETE FROM "LegacyDirectRun"');
  await add({ id: 'pending-without-run', resolved: null, amount: 0 });
  const result = await same();
  assert.equal(result.length, 1); assert.equal(result[0].resolved, false);
  assert.equal(result[0].current.teacherPaySnapshot, 0);
  const fingerprint = result[0].fingerprint;
  await db.exec(`INSERT INTO "LegacyDirectRun" VALUES('co','not-final','running','2026-11-10T00:00:00Z')`);
  assert.equal((await same())[0].fingerprint, fingerprint);
});

test('duplicate CommonModel general ids delegate to original ordering instead of choosing a new ambiguous tie order', async () => {
  await add({ id: 'shared-conflict', dataset: 59 });
  await db.exec(`INSERT INTO "CommonModel" SELECT 'duplicate-map',company,"generalId","modelId","itemId",'另一个学员','另一个课程' FROM "CommonModel"`);
  await trackLegacyCalls();
  // Both original rows share the same conflict id; ORDER BY conflict id alone
  // does not define their relative order. Equality in one plan is insufficient.
  await compareFallback(1);
  const result = await same();
  assert.equal(result.length, 2);
  assert.equal(new Set(result.map(row => row.fingerprint)).size, 2);
});

test('authority drift falls back to the newly authoritative output and reinstall cannot reauthorize stale bulk semantics', async () => {
  await add();
  await db.exec(`CREATE OR REPLACE FUNCTION xs_payroll_cutoff_conflicts(co text,mon text) RETURNS SETOF jsonb LANGUAGE sql AS $$
    SELECT jsonb_build_object('authorityRevision',2,'company',co,'month',mon,'resolved',false,'current',jsonb_build_object('teacherPaySnapshot',73.25))
  $$;`);
  assert.equal((await one('SELECT xs_payroll_read_snapshot_authority_ready() ready')).ready, false);
  assert.deepEqual(await same(), [{ authorityRevision: 2, company: 'co', month, resolved: false, current: { teacherPaySnapshot: 73.25 } }]);
  await db.exec(authorityInstall);
  await db.exec(read('../sql/payroll-cutoff-bulk.sql'));
  assert.equal((await one('SELECT xs_payroll_read_snapshot_authority_ready() ready')).ready, false);
  await same();
});

test('date parser drift delegates to original new date decisions', async () => {
  await add({ id: 'future-under-old-parser', date: '2026-12-01' });
  assert.equal((await same()).length, 0);
  await db.exec(`CREATE OR REPLACE FUNCTION xs_payroll_time(v text) RETURNS timestamptz LANGUAGE sql IMMUTABLE AS $$ SELECT NULL::timestamptz $$;`);
  assert.equal((await one('SELECT xs_payroll_read_snapshot_authority_ready() ready')).ready, false);
  const changed = await same();
  assert.equal(changed.length, 1); assert.equal(changed[0].resolved, false);
});

test('evidence membership preserves textual IDs, numeric grouping, duplicates and missing addon rows',async()=>{
 const original=`WITH local_rows AS MATERIALIZED (SELECT c."generalId" FROM "CommonModel" c JOIN "MemoryPracticeRecord" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=60), wanted AS MATERIALIZED (SELECT DISTINCT jsonb_array_elements_text($2::jsonb) AS id) SELECT count(*)::int value FROM (SELECT l."generalId" FROM local_rows l JOIN wanted w ON w.id=l."generalId"::text GROUP BY l."generalId" HAVING count(*)=1) x`;
 await db.exec(`INSERT INTO "CommonModel"("objectId",company,"generalId","modelId","itemId") VALUES('a','co',1,60,1),('b','co',1.0,60,2),('c','co',2,60,3),('d','co',3,60,99),('e','other',4,60,4);INSERT INTO "MemoryPracticeRecord"(company,id) VALUES('co',1),('co',2),('co',3),('co',3),('other',4);`);
 for(const ids of [['1'],['1.0'],['1','1.0'],['1','1'],['01'],['2','3','4'],[],[null,'1']]){
  const args=['co',JSON.stringify(ids)];const expected=(await one(original,args)).value;
  assert.equal((await one('SELECT xs_payroll_evidence_matches($1,60,$2::jsonb) value',args)).value,expected);
 }
});