xiaoshu-app/scripts/cloud/legacy-direct-control.js

async function directLease(db,p,co) {
 const r=await db.oneOrNone('UPDATE "LegacyDirectRun" SET "leaseUntil"=now()+interval \'5 minutes\' WHERE company=$1 AND "runId"=$2 AND owner=$3 AND status=\'running\' AND "leaseUntil">now() RETURNING "runId"',[co,p.runId,p.owner]);
 if(!r)throw Error('sync_lease_unavailable');
}
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 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={},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(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,includePending=false) {
 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 ($3 OR NOT COALESCE(s."pendingIncome",false)) AND v.status IN ('historical','exception')
  AND (v."incomeMonth">='2026-09' OR xs_legacy_review_needs_fee(m.company,m."objectId"))
  AND r."completionRole"='historical_unknown'`,[co,ids,includePending]);
 const verified=new Set(),zeros=new Map(),trials=new Map(),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(row.plid)!==coach||!fields.every(k=>String(fee[k]??'')===String(row[k]??'')))continue;
    if(Number(fee.money)>0)verified.add(id);
    if(/^0(?:[.]0+)?$/.test(String(fee.money)))zeros.set(id,fee);
   }
  }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}));
  }
 }
 if(zeros.size){
  const zeroRows=rows.filter(row=>zeros.has(Number(row.GeneralID??row.generalId)));
  try{
   const [appointments,studies]=await Promise.all([
    directReviewSourceById(54,zeroRows.map(row=>row.orderId)),
    directReviewSourceById(56,zeroRows.map(row=>row.xxjlid)),
   ]);
   for(const row of zeroRows){
    const id=Number(row.GeneralID??row.generalId),appointment=appointments.get(Number(row.orderId)),study=studies.get(Number(row.xxjlid));
    if(!appointment||String(appointment.dslx)!=='3'||!/[（(]体验课[）)]$/.test(String(appointment.Title??'').trim())||
       String(appointment.szyh)!==String(row.yhid)||!study||String(study.dsid)!==String(row.orderId))continue;
    const pick=(value,keys)=>Object.fromEntries(keys.map(key=>[key,value[key]]));
    trials.set(id,{fee:pick(zeros.get(id),['GeneralID','money',...fields]),
     appointment:pick(appointment,['GeneralID','Title','szyh','dslx']),study:pick(study,['GeneralID','dsid'])});
   }
  }catch{console.warn(JSON.stringify({event:'legacy_review_trial_source_unavailable'}));}
 }
 return rows.map(row=>{
  const id=Number(row.GeneralID??row.generalId);
  return {...row,reviewFeeVerified:verified.has(id),reviewTrialProof:trials.get(id)??null};
 });
}

async function directReviewSourceById(modelId,values) {
 const ids=[...new Set(values.map(Number).filter(id=>Number.isSafeInteger(id)&&id>0))];
 const result=new Map(),deadline=Date.now()+15000;
 for(let i=0;i<ids.length;i+=75){
  const batch=ids.slice(i,i+75),remaining=deadline-Date.now();
  if(remaining<=0)throw Error('legacy_review_source_timeout');
  const {items,page}=await directOld('content_list',{modelId,ids:batch.join(','),cpage:1,psize:100},Math.min(6000,remaining));
  const pages=Number(page.pageCount);
  if(Number(page.itemCount)!==items.length||!Number.isSafeInteger(pages)||pages<0||pages>1)throw Error('legacy_review_source_incomplete');
  for(const item of items){
   const id=Number(item.GeneralID);
   if(!batch.includes(id))throw Error('legacy_review_source_mismatch');
   result.set(id,result.has(id)?null:item);
  }
 }
 return result;
}

async function directPendingTrialProof(co,afterTaskId) {
 // A task can become complete during projection, after its first source-page read.
 const rows=await Psql.query(`SELECT c."generalId" FROM "LegacyReviewSyncState" s
  JOIN "MemoryPracticeRecord" m ON m.company=s.company AND m."objectId"=s."taskId"
  JOIN "CommonModel" c ON c.company=m.company AND c."modelId"=60 AND c."itemId"=m.id
  WHERE s.company=$1 AND s."taskId" IN (
   SELECT "taskId" FROM "LegacyReviewSyncState" WHERE company=$1 AND "pendingIncome" AND "taskId">$2 ORDER BY "taskId" LIMIT 100
  ) AND s."legacyFeeProof" IS NULL AND xs_legacy_review_needs_fee(m.company,m."objectId")`,[co,afterTaskId]);
 if(!rows.length)return [];
 try{
  const source=await directReviewSourceById(60,rows.map(row=>row.generalId));
  return await directReviewFeeProof(co,[...source.values()].filter(Boolean),true);
 }catch{
  console.warn(JSON.stringify({event:'legacy_review_pending_trial_source_unavailable'}));
  return [];
 }
}

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])};
 if(p.action==='claim')return Psql.tx(async t=>{
  await t.any('SELECT pg_advisory_xact_lock(hashtext($1))',['direct-sync:'+co]);
  const today=(await t.one("SELECT to_char(now() AT TIME ZONE 'Asia/Shanghai','YYYY-MM-DD') AS business_day,to_char((now() AT TIME ZONE 'Asia/Shanghai')-interval '1 day','YYYY-MM-DD') AS yesterday"));
  const day=p.targetBusinessDate||today.business_day;if(![today.business_day,today.yesterday].includes(day))throw Error('invalid_business_date');
  const latest=await t.oneOrNone('SELECT * FROM "LegacyDirectRun" WHERE company=$1 AND status=\'running\' ORDER BY "startedAt" DESC LIMIT 1',[co]);
  if(latest&&latest.owner!==p.owner&&new Date(latest.leaseUntil).getTime()>Date.now())throw Error('another_sync_running');
  const existing=latest||await t.oneOrNone('SELECT * FROM "LegacyDirectRun" WHERE company=$1 AND "businessDate"=$2',[co,day]);
  if(existing&&existing.status!=='running')return {alreadyCompleted:true,...existing};
  if(!/^[a-zA-Z0-9-]{10,80}$/.test(p.owner||'')||!/^[a-zA-Z0-9-]{10,80}$/.test(p.proposedRunId||''))throw Error('invalid_run_identity');
  if(existing){await t.none('UPDATE "LegacyDirectRun" SET owner=$2,"leaseUntil"=now()+interval \'5 minutes\' WHERE "runId"=$1',[existing.runId,p.owner]);return {runId:existing.runId,businessDate:existing.businessDate,resumed:true};}
  await t.none('INSERT INTO "LegacyDirectRun" (company,"businessDate","runId",owner,"leaseUntil") VALUES($1,$2,$3,$4,now()+interval \'5 minutes\')',[co,day,p.proposedRunId,p.owner]);return {runId:p.proposedRunId,businessDate:day,resumed:false};
 });
 if(p.action==='finish'){const prior=await Psql.oneOrNone('SELECT status,summary FROM "LegacyDirectRun" WHERE company=$1 AND "runId"=$2 AND owner=$3',[co,p.runId,p.owner]);if(prior&&prior.status!=='running')return prior.summary;}
 await directLease(Psql,p,co);
 if(p.action==='review-income'){
  if(!/^review-income:[A-Za-z0-9:._-]{1,140}$/.test(p.batchKey||''))throw Error('Invalid review income batch');
  const afterTaskId=String(p.afterTaskId||'');
  const proof=await directPendingTrialProof(co,afterTaskId);
  return Psql.tx(async t=>{
   await directLease(t,p,co);
   const fingerprint=(await t.one('SELECT md5($1) value',[JSON.stringify(['review-income',String(p.afterTaskId||'')])])).value;
   const prior=await t.oneOrNone('SELECT fingerprint,result FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 AND key=$3',[co,p.runId,p.batchKey]);if(prior){if(prior.fingerprint!==fingerprint)throw Error('batch_key_reused');return prior.result;}
   if(proof.length){
    await t.none('LOCK TABLE "MemoryPracticeRecord","ReviewReward" IN SHARE ROW EXCLUSIVE MODE');
    await t.one('SELECT xs_legacy_review_observe($1,$2::jsonb)',[co,JSON.stringify(proof)]);
   }
   const result=(await t.one('SELECT xs_legacy_review_income($1,$2,100,$3) result',[co,u.id,afterTaskId])).result;
   await t.none('INSERT INTO "LegacyDirectBatch"(company,"runId",key,fingerprint,result) VALUES($1,$2,$3,$4,$5::jsonb)',[co,p.runId,p.batchKey,fingerprint,JSON.stringify({...result,modelId:0})]);return result;
  });
 }
 if(p.action==='finance-members')return Psql.query('SELECT u."legacyUserId" id FROM "_User" u WHERE u.company=$1 AND u."identityType"=\'member\' AND COALESCE(u."isDeleted",false)=false AND (u."sourceKey" LIKE \'legacy%\' OR u."sourceKey" LIKE \'[["UserID",%\') AND COALESCE(u."legacyUserData"->>\'SyncExcluded\',\'false\')<>\'true\' AND ((SELECT count(*) FROM "LegacyFinanceReceipt" f WHERE f.company=u.company AND f."studentId"=u."legacyUserId" AND f."runId"=$2)<4 OR NOT EXISTS(SELECT 1 FROM "LegacyCreditBalanceReceipt" b WHERE b.company=u.company AND b."studentId"=u."legacyUserId" AND b."runId"=$2)) ORDER BY u."legacyUserId"',[co,p.runId]);
 if(p.action==='finance'){
  const result=await importLegacyFinance(co,Number(p.studentId),p.runId,args=>directOld('user_point_list',args));
  result.balances=await reconcileLegacyCreditBalances(co,Number(p.studentId),p.runId);return result;
 }
 if(p.action==='source'){
  const m=Number(p.modelId);if(!specs[m])throw Error('invalid_model');
  if(p.id){if(!Number.isSafeInteger(p.id)||p.id<1)throw Error('invalid_id');return directOld('content_get',{id:p.id});}
  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(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;
 }
 if(p.action==='supplement')return directSupplement(p,co);
 if(p.action==='conflicts'){
  if(!Array.isArray(p.rows)||p.rows.length>100)throw Error('invalid_conflicts');
  return Psql.tx(async t=>{await directLease(t,p,co);for(const c of p.rows){if(!specs[Number(c.modelId)]||!Number.isSafeInteger(c.generalId))throw Error('invalid_conflict');
   const source='direct-api:'+c.modelId+':'+c.generalId;
   await t.none('INSERT INTO "LegacySyncConflict" ("objectId",company,"sourceKey",dataset,"legacyKey","conflictType",message,resolved,resolution,"createdAt","updatedAt") SELECT substr(md5($1),1,20),$2,$1,$3,$4,\'direct-api-preserve-new\',$5,false,\'保留新系统，待人工复核\',now(),now() WHERE NOT EXISTS(SELECT 1 FROM "LegacySyncConflict" WHERE company=$2 AND "sourceKey"=$1)',[source,co,String(c.modelId),String(c.generalId),JSON.stringify({fields:c.fields||[],reason:c.reason||'保留新系统'})]);
  }return {saved:p.rows.length};});
 }
 if(p.action==='verify'){
  const m=Number(p.modelId),s=specs[m];if(!s||!Array.isArray(p.ids)||p.ids.length>10000||p.ids.some(x=>!Number.isSafeInteger(x)||x<1))throw Error('invalid_verification');
  return Psql.query('SELECT c."generalId",count(*)::int common_rows,count(a."objectId")::int addon_rows FROM "CommonModel" c LEFT JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId"=ANY($3::numeric[]) GROUP BY c."generalId" ORDER BY c."generalId"',[co,m,p.ids]);
 }
 if(p.action==='evidence'){
  const m=Number(p.modelId),s=specs[m];if(!s)throw Error('invalid_evidence_model');
  const latest=await directOld('content_list',{modelId:m,psize:1,cpage:1,...(m===2?{nid:3}:{})});
  return Psql.tx(async t=>{
   await directLease(t,p,co);
   const pages=await t.any('SELECT page,ids,total,checksum FROM "LegacyDirectSourcePage" WHERE company=$1 AND "runId"=$2 AND model=$3 ORDER BY page',[co,p.runId,m]);
   const ids=[...new Set(pages.flatMap(r=>r.ids))].sort((a,b)=>a-b),total=Number(latest.page.itemCount??latest.items.length);
   if(!pages.length||ids.length!==total||pages.some((r,i)=>Number(r.page)!==i+1||Number(r.total)!==total)||pages.length!==Math.max(1,Math.ceil(total/1000)))throw Error('source_shifted_restart_dataset');
   const rows=await t.any('SELECT c."generalId",count(*)::int common_rows,count(a."objectId")::int addon_rows FROM "CommonModel" c LEFT JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId" IN (SELECT jsonb_array_elements_text($3::jsonb)::numeric) GROUP BY c."generalId" ORDER BY c."generalId"',[co,m,JSON.stringify(ids)]);
   if(rows.length!==ids.length||rows.some(r=>r.common_rows!==1||r.addon_rows!==1))throw Error('server_projection_reconciliation_failed');
   const checks=await t.one('SELECT md5($1) source,md5($2) target,md5($3) content',[ids.join('\n'),rows.map(r=>Number(r.generalId)).join('\n'),pages.map(r=>r.checksum).join('\n')]);
   if(checks.source!==checks.target)throw Error('server_id_checksum_failed');
   await t.none('INSERT INTO "LegacyDirectEvidence" (company,"runId",model,ids,"sourceCount","sourceChecksum","targetChecksum",conflicts,"contentChecksum") VALUES($1,$2,$3,$4::jsonb,$5,$6,$7,$8,$9) ON CONFLICT(company,"runId",model) DO UPDATE SET ids=EXCLUDED.ids,"sourceCount"=EXCLUDED."sourceCount","sourceChecksum"=EXCLUDED."sourceChecksum","targetChecksum"=EXCLUDED."targetChecksum",conflicts=EXCLUDED.conflicts,"contentChecksum"=EXCLUDED."contentChecksum"',[co,p.runId,m,JSON.stringify(ids),ids.length,checks.source,checks.target,Number(p.conflicts||0),checks.content]);
   return {modelId:m,sourceCount:ids.length,sourceChecksum:checks.source,targetChecksum:checks.target,sourcePageContentChecksum:checks.content,reconciled:true};
  });
 }
 if(p.action==='reset-dataset'){if(!specs[Number(p.modelId)])throw Error('invalid_model');await Psql.none('DELETE FROM "LegacyDirectSourcePage" WHERE company=$1 AND "runId"=$2 AND model=$3',[co,p.runId,p.modelId]);await Psql.none('DELETE FROM "LegacyDirectEvidence" WHERE company=$1 AND "runId"=$2 AND model=$3',[co,p.runId,p.modelId]);return {reset:true};}
 if(p.action==='finish')return Psql.tx(async t=>{
  await directLease(t,p,co);
  const expected=[2,52,58,56,53,60,54,59,61],d=p.summary?.datasets;
  if(!Array.isArray(d)||d.length!==expected.length||!expected.every(m=>d.some(x=>x.modelId===m&&x.reconciled&&x.failed===0)))throw Error('reconciliation_incomplete');
  // Recompute ID coverage on the server, using the actual run's captured IDs, before a success receipt.
  const evidence=await t.any('SELECT model,ids,"sourceCount","sourceChecksum","targetChecksum" FROM "LegacyDirectEvidence" WHERE company=$1 AND "runId"=$2',[co,p.runId]);
  if(evidence.length!==expected.length)throw Error('server_reconciliation_missing');
  for(const e of evidence){const s=specs[e.model],r=await t.one('SELECT count(*)::int n FROM (SELECT c."generalId" FROM "CommonModel" c JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId" IN (SELECT jsonb_array_elements_text($3::jsonb)::numeric) GROUP BY c."generalId" HAVING count(*)=1) x',[co,e.model,JSON.stringify(e.ids)]);if(r.n!==Number(e.sourceCount)||e.sourceChecksum!==e.targetChecksum)throw Error('server_id_reconciliation_failed');}
  const totals=await t.any('SELECT (result->>\'modelId\')::int model,COALESCE(sum((result->>\'createdCommon\')::int),0)::int created,COALESCE(sum((result->>\'createdAddon\')::int),0)::int addon,COALESCE(sum((result->>\'updated\')::int),0)::int updated FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 GROUP BY (result->>\'modelId\')::int',[co,p.runId]);
  for(const x of d){const total=totals.find(t=>t.model===x.modelId);x.created=total?.created||0;x.createdAddon=total?.addon||0;x.updated=total?.updated||0;}
  const members=await t.any('SELECT "legacyUserId" id FROM "_User" WHERE company=$1 AND "identityType"=\'member\' AND COALESCE("isDeleted",false)=false AND ("sourceKey" LIKE \'legacy%\' OR "sourceKey" LIKE \'[["UserID",%\') AND COALESCE("legacyUserData"->>\'SyncExcluded\',\'false\')<>\'true\'',[co]);
  const finance=await t.any('SELECT "studentId",account,ids,checksum FROM "LegacyFinanceReceipt" WHERE company=$1 AND "runId"=$2',[co,p.runId]);
  if(members.some(m=>finance.filter(f=>Number(f.studentId)===Number(m.id)).length!==4))throw Error('finance_reconciliation_incomplete');
  for(const [account,table] of [['Purse','UserExpDomP'],['SilverCoin','UserSIcon'],['UserExp','UserExpHis'],['UserPoint','UserUserPoint']]){
   const mismatch=await t.one('SELECT count(*)::int n FROM "LegacyFinanceReceipt" r WHERE r.company=$1 AND r."runId"=$2 AND r.account=$3 AND (jsonb_array_length(r.ids)<>(SELECT count(*) FROM "'+table+'" l WHERE l.company=r.company AND l."userId"=r."studentId" AND l."legacyFinanceId" IN (SELECT jsonb_array_elements_text(r.ids))) OR r.checksum<>(SELECT md5(COALESCE(jsonb_agg(jsonb_build_object(\'id\',l."legacyFinanceId",\'hash\',l."legacyFinanceHash") ORDER BY l."legacyFinanceId"::numeric),\'[]\'::jsonb)::text) FROM "'+table+'" l WHERE l.company=r.company AND l."userId"=r."studentId" AND l."legacyFinanceId" IN (SELECT jsonb_array_elements_text(r.ids))))',[co,p.runId,account]);if(mismatch.n)throw Error('finance_postimage_changed');
  }
  const balanceReceipts=await t.any('SELECT "studentId",result FROM "LegacyCreditBalanceReceipt" WHERE company=$1 AND "runId"=$2',[co,p.runId]);
  if(members.some(m=>!balanceReceipts.some(b=>Number(b.studentId)===Number(m.id))))throw Error('balance_reconciliation_incomplete');
  const balances={members:balanceReceipts.length,applied:balanceReceipts.reduce((n,b)=>n+Number(b.result.applied||0),0),conflicts:balanceReceipts.reduce((n,b)=>n+Number(b.result.conflicts||0),0)};
  const summary={...p.summary,finance:{members:members.length,streams:finance.length,rows:finance.reduce((n,f)=>n+f.ids.length,0),checksum:(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(finance.sort((a,b)=>(a.studentId+":"+a.account).localeCompare(b.studentId+":"+b.account)))])).hash,reconciled:true,balances},createdBusinessRecords:d.reduce((n,x)=>n+x.created,0),refreshedBusinessRecords:d.reduce((n,x)=>n+x.updated,0),completedAt:new Date().toISOString(),source:'old-api/WXAPP',bridgeUsed:false,continuousSync:false,settlementReady:false,nightlyAutomation:true};
  const income=await t.one('SELECT COALESCE(sum((result->>\'earned\')::int),0)::int earned,COALESCE(sum((result->>\'excluded\')::int),0)::int excluded,COALESCE(sum((result->>\'amount\')::numeric),0) amount FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 AND result->>\'modelId\'=\'0\'',[co,p.runId]);
  const pending=await t.one('SELECT count(*)::int n FROM "LegacyReviewSyncState" WHERE company=$1 AND "pendingIncome"',[co]);
  summary.reviewIncome={...income,amount:Number(income.amount),pending:pending.n,errors:await t.any('SELECT \"taskId\",\"lastError\" AS reason FROM \"LegacyReviewSyncState\" WHERE company=$1 AND \"pendingIncome\" ORDER BY \"taskId\" LIMIT 25',[co])};
  summary.conflicts=d.reduce((n,x)=>n+Number(x.conflicts||0),0)+Number(summary.accounts?.conflicts||0)+pending.n+balances.conflicts;
  summary.status=summary.conflicts?'completed-with-preserved-conflicts':'completed';
  await t.none('UPDATE "LegacyDirectRun" SET status=$3,summary=$4::jsonb,"finishedAt"=now(),"leaseUntil"=now() WHERE company=$1 AND "runId"=$2',[co,p.runId,summary.status,JSON.stringify(summary)]);
  await t.none('INSERT INTO "SysLog" ("objectId",company,"sourceKey",type1,type2,type3,detail,"createdAt","updatedAt") VALUES(substr(md5($1),1,20),$2,$1,\'migration\',\'legacy-direct-sync-completed\',\'nightly\',$3,now(),now())',['direct-nightly:'+p.runId,co,JSON.stringify({actor:u.id,action:'legacy-direct-sync-completed',after:summary})]);
  return summary;
 });
 throw Error('invalid_control_action');
}