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

async function handler(request,response){
try{
 const u=request.user;if(!u)return response.status(401).json({success:false});await u.fetch({useMasterKey:true});if(u.get('isDisabled')||u.get('isDeleted')||u.get('adminRoleKey')!=='super-admin'||u.get('company')?.id!==__COMPANY_ID__)return response.status(403).json({success:false});
 const co=__COMPANY_ID__,p=request.body?.params||{},specs=__SPECS__,windows=__WINDOWS__,q=x=>'"'+x+'"',s=specs[p.modelId],m=Number(p.modelId);
 if(p.action==='status'||p.action==='claim'||p.action==='source'||p.action==='finish'||p.action==='conflicts'||p.action==='verify'||p.action==='supplement'||p.action==='evidence'||p.action==='reset-dataset')return response.json({success:true,data:await directControl(p,co,u)});
 await directLease(Psql,p,co);
 const legacy=r=>r&&typeof r.sourceKey==='string'&&(/\[\["(?:ID|GeneralID)",\d+\]\]/.test(r.sourceKey)||/^legacy:model:\d+:general:\d+$/.test(r.sourceKey));
 const unchanged=r=>legacy(r)&&!r.teachingManaged&&(r.createdAt===r.updatedAt||!r.updatedAt||Date.parse(r.updatedAt)<Date.parse('2026-08-01')||windows.some(([a,b])=>Date.parse(r.updatedAt)>=a&&Date.parse(r.updatedAt)<=b));
 const sameVersion=(a,b)=>a===b;
 if(p.action==='overview'){
 const data=[];for(const [id,spec] of Object.entries(specs)){const row=await Psql.one('SELECT (SELECT count(*)::int FROM "CommonModel" WHERE company=$1 AND "modelId"=$2) common,(SELECT count(*)::int FROM "'+spec.className+'" WHERE company=$1) addon,(SELECT count(*)::int FROM "CommonModel" c WHERE c.company=$1 AND c."modelId"=$2 AND NOT EXISTS(SELECT 1 FROM "'+spec.className+'" a WHERE a.company=c.company AND a.id=c."itemId")) missing_links',[co,Number(id)]);data.push({modelId:Number(id),...row});}return response.json({success:true,data});
 }
 if(!s)throw Error('Invalid model');
 const readSql='SELECT c."generalId",to_jsonb(c) AS common,to_jsonb(a) AS addon'+(m===60?',to_jsonb(r) AS reward':'')+' FROM "CommonModel" c LEFT JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId"'+(m===60?' LEFT JOIN "ReviewReward" r ON r.company=a.company AND r."taskId"=a."objectId"':'')+' WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId"=ANY($3::numeric[])';
 if(p.action==='read'){const ids=p.ids||[];if(ids.length>1000||ids.some(x=>!Number.isSafeInteger(x)||x<1))throw Error('Invalid IDs');return response.json({success:true,data:await Psql.query(readSql,[co,m,ids])});}
 if(!['append','patch'].includes(p.action)||!Array.isArray(p.rows)||p.rows.length>50)throw Error('Invalid mutation');
 const result=await Psql.tx(async t=>{
  await directLease(t,p,co);
  const fingerprint=(await t.one('SELECT md5($1) fingerprint',[JSON.stringify([p.action,m,p.rows])])).fingerprint;
  const receipt=await t.oneOrNone('SELECT fingerprint,result FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 AND key=$3',[co,p.runId,p.batchKey]);if(receipt){if(receipt.fingerprint!==fingerprint)throw Error('batch_key_reused');return receipt.result;}
  await t.any("SELECT pg_advisory_xact_lock(hashtext('xs-credit:'||$1))",[co]);
  const out={modelId:m,createdCommon:0,createdAddon:0,updated:0,duplicates:0,rewardEvidence:0,conflicts:[],postimages:[]};
  async function insert(table,data,allowed){if(Object.keys(data).some(k=>!allowed.includes(k)))throw Error('Unexpected projected field');const cols=Object.keys(data).map(q).join(',');return t.one('INSERT INTO '+q(table)+' ('+cols+') SELECT '+Object.keys(data).map(k=>'v.'+q(k)).join(',')+' FROM jsonb_populate_record(NULL::'+q(table)+',$1::jsonb) v RETURNING *',[JSON.stringify(data)]);}
  for(const incoming of p.rows){
   const gid=Number(incoming.generalId);if(!Number.isSafeInteger(gid)||gid<1)throw Error('Invalid incoming ID');
   const cc=(await t.oneOrNone('SELECT to_jsonb(c) data FROM "CommonModel" c WHERE company=$1 AND "modelId"=$2 AND "generalId"=$3 FOR UPDATE',[co,m,gid]))?.data;
   if(p.action==='append'){
    if(cc&&(cc.teachingManaged||Number(cc.status)===-2||!legacy(cc))){out.conflicts.push({generalId:gid,reason:'新系统课次已受保护'});continue;}
    let item=cc?Number(cc.itemId):Number(incoming.addon?.id||gid);
    let aa=(await t.oneOrNone('SELECT to_jsonb(a) data FROM '+q(s.className)+' a WHERE company=$1 AND id=$2 FOR UPDATE',[co,item]))?.data;
    if(cc&&aa)continue;
    const sourceKey='legacy:model:'+m+':general:'+gid;
    if(!cc&&aa){item=900000000000+gid*100+m;aa=(await t.oneOrNone('SELECT to_jsonb(a) data FROM '+q(s.className)+' a WHERE company=$1 AND id=$2 FOR UPDATE',[co,item]))?.data;if(aa){out.conflicts.push({generalId:gid,reason:'附表编号冲突'});continue;}}
    const orphan=await t.oneOrNone('SELECT to_jsonb(a) data FROM '+q(s.className)+' a WHERE company=$1 AND "sourceKey"=$2 FOR UPDATE',[co,sourceKey]);
    if(orphan){out.conflicts.push({generalId:gid,reason:'已有同源孤立附表，保留待核对'});continue;}
    const now=new Date().toISOString(),common={...incoming.common,company:co,sourceKey,generalId:gid,itemId:item,modelId:m,createdAt:now,updatedAt:now},addon={...incoming.addon,company:co,sourceKey,id:item,createdAt:now,updatedAt:now};
    if(!common.objectId||!addon.objectId)throw Error('Missing object IDs');
    if(m===60){
     const key=(await t.one('SELECT xiaoshu_review_key($1,$2,$3) key',[String(addon.yhid),String(addon.xxjlid),String(addon.kywrq).slice(0,10)])).key;
     const duplicate=key&&await t.oneOrNone('SELECT "objectId" FROM "MemoryPracticeRecord" WHERE company=$1 AND "rewardTaskKey"=$2 LIMIT 1',[co,key]);
     addon.rewardTaskKey=duplicate?null:key;addon.rewardDuplicate=Boolean(duplicate);if(duplicate)out.duplicates++;
     // Same bounded historical-import semantics used on 2026-10-08. Only this transaction and this allowlisted table; restore triggers before recording historical zero-amount evidence.
     await t.none("SET LOCAL session_replication_role='replica'");
     await insert(s.className,addon,s.addonColumns);await t.none("SET LOCAL session_replication_role='origin'");
     if(String(addon.fxzt)==='1'){await t.any('SELECT xiaoshu_record_review_reward(m,true) FROM "MemoryPracticeRecord" m WHERE m.company=$1 AND m."objectId"=$2',[co,addon.objectId]);out.rewardEvidence++;}
    }else await insert(s.className,addon,s.addonColumns);
    out.createdAddon++;
    if(!cc){await insert('CommonModel',common,s.commonColumns);out.createdCommon++;}
   }else{
    if(!cc){out.conflicts.push({generalId:gid,reason:'主记录不存在'});continue;}
    const aa=(await t.oneOrNone('SELECT to_jsonb(a) data FROM '+q(s.className)+' a WHERE company=$1 AND id=$2 FOR UPDATE',[co,cc.itemId]))?.data;
    if(!aa||cc.objectId!==incoming.commonId||aa.objectId!==incoming.addonId||!sameVersion(cc.updatedAt,incoming.commonVersion)||!sameVersion(aa.updatedAt,incoming.addonVersion)||!(await directUnchanged(t,co,m,gid,cc,aa,unchanged))){
     out.conflicts.push({generalId:gid,reason:'新系统已有修改、版本变化或来源不可确认'});continue;
    }
    if(m===2||m===52||cc.teachingManaged||aa.teachingManaged||m===54&&(aa.bindingId||aa.creditHoldStatus||aa.courseCategoryKey)||m===58&&aa.enrollmentStatus==='active'){out.conflicts.push({generalId:gid,reason:'保留新系统课程内容或教学凭据'});continue;}
    const fields=Object.keys(incoming.values||{});if(!fields.length||fields.some(f=>!s.fields.includes(f)))throw Error('Invalid patch fields');
    if(m===60&&(Number(aa.assignmentVersion)>0||await t.oneOrNone('SELECT "objectId" FROM "ReviewReward" WHERE company=$1 AND "taskId"=$2',[co,aa.objectId])||fields.some(f=>['yhid','xxjlid','kywrq','plid'].includes(f)))){out.conflicts.push({generalId:gid,reason:'保留奖励凭据或复习身份'});continue;}
    if(m===60)await t.none("SET LOCAL session_replication_role='replica'");
    await t.none('UPDATE '+q(s.className)+' a SET '+fields.map(f=>q(f)+'=v.'+q(f)).join(',')+',"updatedAt"=now() FROM jsonb_populate_record(NULL::'+q(s.className)+',$3::jsonb) v WHERE a.company=$1 AND a."objectId"=$2',[co,aa.objectId,JSON.stringify(incoming.values)]);
    if(m===60){await t.none("SET LOCAL session_replication_role='origin'");await t.any('SELECT xiaoshu_record_review_reward(m,true) FROM "MemoryPracticeRecord" m WHERE m.company=$1 AND m."objectId"=$2 AND m.fxzt::text=\'1\'',[co,aa.objectId]);}
    out.updated++;
   }
   const after=await t.oneOrNone(readSql,[co,m,[gid]]);if(!after?.common||!after.addon)throw Error('Atomic projection verification failed');
   if(p.action==='patch'){const expected=(await t.one('SELECT to_jsonb(v) expected FROM jsonb_populate_record(NULL::'+q(s.className)+',$1::jsonb) v',[JSON.stringify(incoming.values)])).expected;if(Object.keys(incoming.values).some(f=>JSON.stringify(expected[f])!==JSON.stringify(after.addon[f])))throw Error('Post-write field verification failed');}
   if(m===60&&after.reward&&!(Number(after.reward.amount)===0&&['historical','exception'].includes(after.reward.status)))throw Error('Historical reward verification failed');
   await directBaseline(t,co,m,gid,after);
   out.postimages.push(after);
  }
  if(!/^[A-Za-z0-9:._-]{1,160}$/.test(p.batchKey||''))throw Error('Invalid batch key');
  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(out)]);
  return out;
 });
 return response.json({success:true,data:result});
}catch(e){return response.status(500).json({success:false,message:String(e.message||e)});}
}