fix(sync): recover verified legacy review fees into the queue

scripts/cloud/legacy-direct-control.js

@@ -6,19 +6,70 @@ async function directBaseline(t,co,m,gid,row) {
  await t.none('INSERT INTO "LegacyDirectBaseline" (company,model,gid,fingerprint) VALUES($1,$2,$3,md5($4::jsonb::text)) ON CONFLICT(company,model,gid) DO UPDATE SET fingerprint=EXCLUDED.fingerprint,"updatedAt"=now()',[co,m,gid,JSON.stringify([row.common,row.addon])]);
 }
 async function directUnchanged(t,co,m,gid,c,a,fallback) {
- const r=await t.oneOrNone('SELECT fingerprint=md5($4::jsonb::text) matches FROM "LegacyDirectBaseline" WHERE company=$1 AND model=$2 AND gid=$3',[co,m,gid,JSON.stringify([c,a])]);
+ const snapshot=JSON.stringify([c,a]);
+ let priorSnapshot=null;
+ if(m===60&&a.assignmentVersion===null){
+  const {assignmentVersion,...legacyAddon}=a;
+  priorSnapshot=JSON.stringify([c,legacyAddon]);
+ }
+ const r=await t.oneOrNone('SELECT (fingerprint=md5($4::jsonb::text) OR COALESCE(fingerprint=md5($5::jsonb::text),false)) matches FROM "LegacyDirectBaseline" WHERE company=$1 AND model=$2 AND gid=$3',[co,m,gid,snapshot,priorSnapshot]);
  return r?r.matches:fallback(c)&&fallback(a);
 }
-async function directOld(action,args={}) {
+async function directOld(action,args={},timeout=60000) {
  const config=await Parse.Config.get({useMasterKey:true}),url=config.get('legacyScheduleApiUrl'),apiId=config.get('legacyScheduleApiId'),apiKey=config.get('legacyScheduleApiKey');
  if(!url||!apiId||!apiKey)throw Error('direct_api_not_configured');
  const query=Object.entries({action,...args,apiId,apiKey}).map(([k,v])=>encodeURIComponent(k)+'='+encodeURIComponent(String(v))).join('&');
  let j;
- try {const r=await fetch(url+(url.includes('?')?'&':'?')+query,{signal:AbortSignal.timeout(60000)});if(!r.ok)throw Error();j=await r.json();} catch {throw Error('legacy_read_failed');}
+ try {const r=await fetch(url+(url.includes('?')?'&':'?')+query,{signal:AbortSignal.timeout(timeout)});if(!r.ok)throw Error();j=await r.json();} catch {throw Error('legacy_read_failed');}
  if(Number(j.retcode)===-1)throw Error('legacy_read_rejected');
  const items=typeof j.result==='string'?JSON.parse(j.result):j.result;
  if(!Array.isArray(items))throw Error('legacy_invalid_result');return {items,page:j.page||{}};
 }
+async function directReviewFeeProof(co,rows) {
+ const ids=rows.filter(row=>String(row.fxzt)==='1').map(row=>Number(row.GeneralID??row.generalId));
+ if(!ids.length)return rows;
+ const candidates=await Psql.query(`SELECT c."generalId",r."coachId"
+  FROM "CommonModel" c JOIN "MemoryPracticeRecord" m ON m.company=c.company AND m.id=c."itemId"
+  JOIN "ReviewReward" r ON r.company=m.company AND r."taskId"=m."objectId"
+  JOIN "ReviewRewardIncome" v ON v.company=r.company AND v."objectId"=r."objectId"
+  LEFT JOIN "LegacyReviewSyncState" s ON s.company=m.company AND s."taskId"=m."objectId"
+  WHERE c.company=$1 AND c."modelId"=60 AND c."generalId"=ANY($2::numeric[]) AND m.fxzt='1'
+  AND NOT COALESCE(s."pendingIncome",false) AND v.status IN ('historical','exception')
+  AND v."incomeMonth">='2026-09' AND r."completionRole"='historical_unknown'`,[co,ids]);
+ const verified=new Set(),fields=['yhid','plid','xxjlid','orderId','kcid','fxzt','wcsj','kywrq','kywsj'];
+ const deadline=Date.now()+15000;
+ for(const coach of new Set(candidates.map(row=>Number(row.coachId)))){
+  const coachIds=new Set(candidates.filter(row=>Number(row.coachId)===coach).map(row=>Number(row.generalId)));
+  const income=new Map();
+  let total, count, read=0;
+  try{
+   for(let page=1;;page++){
+    const remaining=deadline-Date.now();
+    if(remaining<=0)throw Error('legacy_review_income_proof_timeout');
+    const result=await directOld('e_get_21list',{uid:coach,status:1,shouru:1,cpage:page,psize:1000},Math.min(6000,remaining));
+    const items=Number(result.page.itemCount),pages=Math.max(1,Number(result.page.pageCount));
+    total??=items;count??=pages;
+    if(!Number.isSafeInteger(total)||!Number.isSafeInteger(count)||count>100||items!==total||pages!==count)throw Error('legacy_review_income_page_changed');
+    read+=result.items.length;
+    for(const row of result.items){
+     const id=Number(row.GeneralID??row.generalId);
+     if(coachIds.has(id))income.set(id,income.has(id)?null:row);
+    }
+    if(page>=count)break;
+   }
+   if(read!==total)throw Error('legacy_review_income_page_incomplete');
+   for(const row of rows){
+    const id=Number(row.GeneralID??row.generalId),fee=income.get(id);
+    if(fee&&Number(fee.money)>0&&Number(row.plid)===coach&&fields.every(k=>String(fee[k]??'')===String(row[k]??'')))verified.add(id);
+   }
+  }catch {
+   // The old fee API can fail for individual accounts. Keep syncing the source page and retry on the next run.
+   console.warn(JSON.stringify({event:'legacy_review_income_proof_unavailable',coach}));
+  }
+ }
+ return rows.map(row=>({...row,reviewFeeVerified:verified.has(Number(row.GeneralID??row.generalId))}));
+}
+
 async function directControl(p,co,u) {
  const specs=__SPECS__;
  if(p.action==='status')return {version:'2026-11-02-review-source-v1',reviewSyncVersion:'source-owned-v1',source:'old-api/WXAPP',bridgeUsed:false,runs:await Psql.query('SELECT "businessDate","runId",status,"startedAt","finishedAt",summary FROM "LegacyDirectRun" WHERE company=$1 ORDER BY "startedAt" DESC LIMIT 3',[co])};
@@ -55,7 +106,7 @@ async function directControl(p,co,u) {
   if(!Number.isSafeInteger(p.page)||p.page<1||p.page>10000)throw Error('invalid_page');
   const result=await directOld('content_list',{modelId:m,psize:p.countOnly?1:1000,cpage:p.page,...(m===2?{nid:3}:{})});
   if(!p.countOnly){const ids=result.items.map(x=>Number(x.GeneralID??x.generalId));if(ids.some(id=>!Number.isSafeInteger(id)||id<1))throw Error('invalid_source_id');
-   if(m===60)await Psql.one('SELECT xs_legacy_review_observe($1,$2::jsonb)',[co,JSON.stringify(result.items)]);
+   if(m===60)await Psql.one('SELECT xs_legacy_review_observe($1,$2::jsonb)',[co,JSON.stringify(await directReviewFeeProof(co,result.items))]);
    await Psql.none('INSERT INTO "LegacyDirectSourcePage" (company,"runId",model,page,ids,total,checksum) VALUES($1,$2,$3,$4,$5::jsonb,$6,md5($7::jsonb::text)) ON CONFLICT(company,"runId",model,page) DO UPDATE SET ids=EXCLUDED.ids,total=EXCLUDED.total,checksum=EXCLUDED.checksum',[co,p.runId,m,p.page,JSON.stringify(ids),Number(result.page.itemCount??ids.length),JSON.stringify(result.items)]);
   }return result;
  }

scripts/sql/legacy-review-sync.sql

@@ -39,25 +39,51 @@ BEGIN
  ON CONFLICT(company,model,gid) DO UPDATE SET fingerprint=EXCLUDED.fingerprint,"updatedAt"=now();
 END $$;
 
--- Bootstrap only records whose complete business fields still match the live old source.
--- Once a baseline exists, an unexplained local modification can never be re-adopted.
+-- Bootstrap exact old-source matches. A completed task can enter the income
+-- queue without becoming writable by the legacy source.
 CREATE OR REPLACE FUNCTION xs_legacy_review_observe(co text, source_rows jsonb) RETURNS integer LANGUAGE plpgsql AS $$
 DECLARE row_data record; n integer:=0;
 BEGIN
  IF jsonb_typeof(source_rows)<>'array' OR jsonb_array_length(source_rows)>1000 THEN RAISE EXCEPTION '无效旧复习来源页'; END IF;
- FOR row_data IN SELECT m."objectId" FROM jsonb_array_elements(source_rows) s
+ FOR row_data IN SELECT m."objectId",m.fxzt,md5(to_jsonb(m)::text) AS fingerprint FROM jsonb_array_elements(source_rows) s
   JOIN "CommonModel" c ON c.company=co AND c."modelId"=60 AND c."generalId"=(s->>'GeneralID')::numeric
   JOIN "MemoryPracticeRecord" m ON m.company=c.company AND m.id=c."itemId"
   LEFT JOIN "LegacyDirectBaseline" b ON b.company=c.company AND b.model=60 AND b.gid=c."generalId"
-  WHERE m.fxzt='0' AND COALESCE(m."assignmentVersion",0)=0 AND m."sourceKey" LIKE 'legacy:model:60:general:%'
+  LEFT JOIN "LegacyReviewSyncState" st ON st.company=m.company AND st."taskId"=m."objectId"
+  WHERE COALESCE(m."assignmentVersion",0)=0 AND m."sourceKey" LIKE 'legacy:model:60:general:%'
   AND NOT COALESCE(c."teachingManaged",false) AND COALESCE(c.status,99)<>-2
-  AND NOT EXISTS(SELECT 1 FROM "ReviewReward" r WHERE r.company=m.company AND r."taskId"=m."objectId")
-  AND NOT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" t WHERE t.company=m.company AND t."taskId"=m."objectId")
-  AND (b.fingerprint IS NULL OR b.fingerprint=md5(jsonb_build_array(to_jsonb(c),to_jsonb(m))::text))
+  AND (b.fingerprint IS NULL OR b.fingerprint=md5(jsonb_build_array(to_jsonb(c),to_jsonb(m))::text)
+       OR (m."assignmentVersion" IS NULL AND b.fingerprint=md5(jsonb_build_array(to_jsonb(c),to_jsonb(m)-'assignmentVersion')::text)))
   AND jsonb_build_array(m.yhid::text,m.plid::text,m.xxjlid::text,m."orderId"::text,m.kcid::text,COALESCE(m.fxzt,''),COALESCE(m.wcsj,''),COALESCE(m.kywrq,''),COALESCE(m.kywsj,''))=
       jsonb_build_array(s->>'yhid',s->>'plid',s->>'xxjlid',s->>'orderId',s->>'kcid',COALESCE(s->>'fxzt',''),COALESCE(s->>'wcsj',''),COALESCE(s->>'kywrq',''),COALESCE(s->>'kywsj',''))
+  AND ((m.fxzt='0' AND st."taskId" IS NULL AND NOT EXISTS(
+         SELECT 1 FROM "ReviewReward" r WHERE r.company=m.company AND r."taskId"=m."objectId"))
+       OR (m.fxzt='1' AND COALESCE(c.status,99)=99
+         AND s->>'reviewFeeVerified'='true'
+         AND m."sourceKey"='legacy:model:60:general:'||c."generalId"::text AND c."sourceKey"=m."sourceKey"
+         AND (st."taskId" IS NULL OR (NOT st."pendingIncome" AND st."assignmentVersion"=0 AND st.fingerprint=md5(to_jsonb(m)::text)))
+         AND EXISTS(SELECT 1 FROM "ReviewReward" r
+           JOIN "ReviewRewardIncome" v ON v.company=r.company AND v."objectId"=r."objectId"
+           WHERE r.company=m.company AND r."taskId"=m."objectId" AND r.amount=0
+           AND v.status IN ('historical','exception') AND v.amount=0 AND v."incomeMonth">='2026-09'
+           AND r."completionRole"='historical_unknown' AND NULLIF(r."completedBy",'') IS NULL
+           AND r."studentId"::text=m.yhid::text AND r."coachId"::text=m.plid::text AND r."plannedDate"=LEFT(m.kywrq,10)
+           AND r."completedAt"=xiaoshu_safe_shanghai_timestamp(m.wcsj) AND r."completedAt"<=now()
+           AND NOT EXISTS(SELECT 1 FROM "PayrollLine" l WHERE l.company=r.company AND l."reviewRewardId"=r."objectId" AND NOT COALESCE(l.excluded,false)))))
   FOR UPDATE OF m,c
- LOOP PERFORM xs_legacy_review_track(co,row_data."objectId");n:=n+1;END LOOP;
+ LOOP
+  IF row_data.fxzt='0' THEN
+   PERFORM xs_legacy_review_track(co,row_data."objectId");
+   n:=n+1;
+  ELSE
+   INSERT INTO "LegacyReviewSyncState"(company,"taskId","assignmentVersion",fingerprint,"pendingIncome")
+   VALUES(co,row_data."objectId",0,row_data.fingerprint,true)
+   ON CONFLICT(company,"taskId") DO UPDATE SET "pendingIncome"=true,"updatedAt"=now()
+   WHERE NOT "LegacyReviewSyncState"."pendingIncome" AND "LegacyReviewSyncState"."assignmentVersion"=0
+     AND "LegacyReviewSyncState".fingerprint=EXCLUDED.fingerprint;
+   IF FOUND THEN n:=n+1;END IF;
+  END IF;
+ END LOOP;
  RETURN n;
 END $$;
 REVOKE ALL ON FUNCTION xs_legacy_review_observe(text,jsonb) FROM PUBLIC;
@@ -74,7 +100,9 @@ BEGIN
  SELECT to_jsonb(t) INTO c FROM "CommonModel" t WHERE t.company=co AND t."modelId"=60 AND t."itemId"=m.id FOR UPDATE;
  IF c IS NULL OR COALESCE((c->>'teachingManaged')::boolean,false) OR c->>'status'='-2' THEN RETURN jsonb_build_object('applied',false,'reason','来源主表受保护'); END IF;
  SELECT fingerprint INTO baseline FROM "LegacyDirectBaseline" WHERE company=co AND model=60 AND gid=(c->>'generalId')::numeric;
- IF (baseline IS NOT NULL AND baseline<>md5(jsonb_build_array(c,before_row)::text)) OR
+ -- Older fingerprints predate assignmentVersion. Only its absent/null form is compatible.
+ IF (baseline IS NOT NULL AND baseline<>md5(jsonb_build_array(c,before_row)::text)
+     AND NOT (m."assignmentVersion" IS NULL AND baseline=md5(jsonb_build_array(c,before_row-'assignmentVersion')::text))) OR
  (baseline IS NULL AND (m."createdAt" IS NULL OR c->>'createdAt' IS NULL OR m."createdAt" IS DISTINCT FROM m."updatedAt" OR c->>'createdAt' IS DISTINCT FROM c->>'updatedAt')) THEN
   RETURN jsonb_build_object('applied',false,'reason','新系统已有修改或来源基线不可确认');
  END IF;

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

@@ -29,7 +29,9 @@ before(async()=>{
  }
  await db.exec('ALTER TABLE "MemoryPracticeRecord" ADD "rewardTaskKey" text, ADD "rewardDuplicate" boolean, ADD "assignmentVersion" numeric DEFAULT 0');
  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(rewardSource.match(/CREATE OR REPLACE FUNCTION xiaoshu_safe_shanghai_timestamp[\s\S]*?\$\$;/)[0]);
- await db.exec('CREATE TABLE "ReviewReward" ("objectId" text,company text,"taskId" text,amount numeric,status text)');
+ await db.exec(`CREATE TABLE "ReviewReward" ("objectId" text,company text,"taskId" text,amount numeric,status text,
+  "coachId" numeric,"studentId" numeric,"plannedDate" text,"completedAt" timestamptz,"completionRole" text,"completedBy" text);
+  CREATE TABLE "PayrollLine" (company text,"reviewRewardId" text,excluded boolean);`);
  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)');
  await db.exec('CREATE VIEW "ReviewRewardIncome" AS SELECT r.*,NULL::text AS "incomeMonth" FROM "ReviewReward" r');
  await db.exec("CREATE FUNCTION xs_review_target(text,text) RETURNS void LANGUAGE plpgsql AS $$ BEGIN RETURN; END $$");
@@ -206,7 +208,7 @@ test('只有全部数据集对账后才能完成，统计来自写入回执，
  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 source=await call({action:'source',modelId,page:1});assert.equal(source.success,true,source.message);
    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});
   }

scripts/tests/legacy-review-sync-policy.test.mjs

@@ -110,6 +110,41 @@ test('历史未初始化任务只在完整业务字段与旧来源一致时建
  const id2=await task('known-local');const source2={...await snapshot('known-local'),GeneralID:id2};await q('DELETE FROM "LegacyReviewSyncState" WHERE "taskId"=\'known-local\'');await q('UPDATE "MemoryPracticeRecord" SET "updatedAt"=\'2026-10-08\' WHERE "objectId"=\'known-local\'');
  assert.equal((await one("SELECT xs_legacy_review_observe('c',$1::jsonb) n",[JSON.stringify([source2])])).n,0);assert.equal((await patch('known-local',{plid:11})).applied,false);
 });
+test('已完成历史凭据漏进队列时，以完整旧来源补入计薪队列，保留原凭据和基线',async()=>{
+ const id=await task('completed-untracked');
+ await patch('completed-untracked',{fxzt:'1',wcsj:'2026-09-05 10:00'});
+ await q(`DELETE FROM "LegacyReviewSyncState" WHERE "taskId"='completed-untracked'`);
+ const before=await snapshot('completed-untracked');
+ const raw=await one(`SELECT to_jsonb(r) row FROM "ReviewReward" r WHERE r."taskId"='completed-untracked'`);
+ const baseline=await one(`SELECT fingerprint FROM "LegacyDirectBaseline" WHERE company='c' AND model=60 AND gid=$1`,[id]);
+ const source={...before,GeneralID:id};
+ const observe=async row=>(await one("SELECT xs_legacy_review_observe('c',$1::jsonb) n",[JSON.stringify([row])])).n;
+ assert.equal(await observe(source),0);
+ assert.equal(await observe({...source,reviewFeeVerified:true,wcsj:'2026-09-05 10:01'}),0);
+ assert.equal(await observe({...source,reviewFeeVerified:true}),1);
+ assert.equal(await observe({...source,reviewFeeVerified:true}),0);
+ assert.equal((await one(`SELECT "pendingIncome" FROM "LegacyReviewSyncState" WHERE "taskId"='completed-untracked'`)).pendingIncome,true);
+ assert.deepEqual(await snapshot('completed-untracked'),before);
+ assert.deepEqual(await one(`SELECT fingerprint FROM "LegacyDirectBaseline" WHERE company='c' AND model=60 AND gid=$1`,[id]),baseline);
+ assert.equal((await drain()).earned,1);
+ assert.equal(Number((await income('completed-untracked')).amount),1);
+ assert.deepEqual(await one(`SELECT to_jsonb(r) row FROM "ReviewReward" r WHERE r."taskId"='completed-untracked'`),raw);
+});
+test('旧基线只缺后来新增的空转交版本字段时允许同步，其他本地差异仍受保护',async()=>{
+ for(const [name,modified] of [['pre-version',false],['pre-version-changed',true]]){
+  const id=await task(name);
+  await db.transaction(async t=>{
+   await t.exec("SET LOCAL xiaoshu.review_transfer='fixture'");
+   await t.query(`UPDATE "MemoryPracticeRecord" SET "assignmentVersion"=NULL WHERE "objectId"=$1`,[name]);
+  });
+  await q(`UPDATE "LegacyDirectBaseline" b SET fingerprint=md5(jsonb_build_array(to_jsonb(c),to_jsonb(m)-'assignmentVersion')::text)
+   FROM "CommonModel" c JOIN "MemoryPracticeRecord" m ON m.company=c.company AND m.id=c."itemId"
+   WHERE b.company=c.company AND b.model=60 AND b.gid=c."generalId" AND b.gid=$1 AND c."modelId"=60`,[id]);
+  if(modified)await q(`UPDATE "MemoryPracticeRecord" SET kywsj='本地变更' WHERE "objectId"=$1`,[name]);
+  assert.equal((await patch(name,{fxzt:'1',wcsj:'2026-09-05 10:00'})).applied,!modified);
+ }
+ assert.equal((await drain()).earned,1);
+});
 test('明确自习老师账号的旧来源只保留进度与核查原因，不自动产生老师工资',async()=>{
  await q(`INSERT INTO "_User"("objectId",company,"legacyUserId","legacyGroupId","identityType",nickname,"legacyUserData") VALUES('study-coach','c',13,3,'coach','自习账号','{}')`);
  await task('old-self');await patch('old-self',{plid:13,fxzt:'1',wcsj:'2026-09-05 10:00'});const result=await drain();assert.equal(result.earned,0);assert.equal(result.pending,1);assert.match(result.errors[0].reason,/自习/);assert.equal(Number((await income('old-self')).amount),0);

scripts/tests/legacy-review-sync-runner.test.mjs

@@ -1,6 +1,32 @@
 import test from 'node:test';import assert from 'node:assert/strict';
 import {mkdtemp,rm} from 'node:fs/promises';import {tmpdir} from 'node:os';import {join} from 'node:path';
+import {readFileSync} from 'node:fs';
+import vm from 'node:vm';
 import {runDirectSync} from '../run-nightly-direct-sync.mjs';
+
+test('旧费用页完整且逐字段匹配的正金额记录才允许补进队列',async()=>{
+ const control=readFileSync(new URL('../cloud/legacy-direct-control.js',import.meta.url),'utf8');
+ const body=control.slice(control.indexOf('async function directReviewFeeProof('),control.indexOf('async function directControl('));
+ const rows=Array.from({length:5},(_,index)=>({
+  GeneralID:index+1,plid:index===4?31:30,yhid:20,xxjlid:90,orderId:80,kcid:70,fxzt:'1',wcsj:'2026-10-13 10:00',
+  kywrq:'2026-10-13',kywsj:'2026-10-13 09:00',reviewFeeVerified:true,
+ }));
+ const warnings=[];
+ const run=vm.runInNewContext(body+';directReviewFeeProof',{
+  Psql:{query:async()=>rows.map(row=>({generalId:row.GeneralID,coachId:row.plid}))},
+  console:{warn:message=>warnings.push(message)},
+  directOld:async(action,{uid,cpage})=>{
+   assert.equal(action,'e_get_21list');
+   if(uid===31)return {items:[{...rows[4],money:1}],page:{pageCount:1,itemCount:2}};
+   return cpage===1?{items:[{...rows[0],money:1},{...rows[1],money:0},{...rows[2],money:1}],page:{pageCount:2,itemCount:5}}:
+    {items:[{...rows[2],money:1},{...rows[3],money:1,wcsj:'2026-10-13 10:01'}],page:{pageCount:2,itemCount:5}};
+  },
+ });
+ const result=await run('c',rows);
+ assert.deepEqual(Array.from(result,row=>row.reviewFeeVerified),[true,false,false,false,false]);
+ assert.ok(rows.every(row=>row.reviewFeeVerified));
+ assert.equal(warnings.length,1);
+});
 test('所有课次同步后分页核定工资，待核定原因进入回执，写入重试复用同一批次',async()=>{
  const root=await mkdtemp(join(tmpdir(),'review-sync-runner-')),original=globalThis.fetch,order=[],receipts=new Map();let simulatedLostResponse=false,applied=0,final;
  try{