legacy-direct-sync.js 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475
  1. async function handler(request,response){
  2. try{
  3. 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});
  4. const co=__COMPANY_ID__,p=request.body?.params||{},specs=__SPECS__,windows=__WINDOWS__,q=x=>'"'+x+'"',s=specs[p.modelId],m=Number(p.modelId);
  5. 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)});
  6. await directLease(Psql,p,co);
  7. const legacy=r=>r&&typeof r.sourceKey==='string'&&(/\[\["(?:ID|GeneralID)",\d+\]\]/.test(r.sourceKey)||/^legacy:model:\d+:general:\d+$/.test(r.sourceKey));
  8. 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));
  9. const sameVersion=(a,b)=>a===b;
  10. if(p.action==='overview'){
  11. 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});
  12. }
  13. if(!s)throw Error('Invalid model');
  14. 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[])';
  15. 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])});}
  16. if(!['append','patch'].includes(p.action)||!Array.isArray(p.rows)||p.rows.length>50)throw Error('Invalid mutation');
  17. const result=await Psql.tx(async t=>{
  18. await directLease(t,p,co);
  19. const fingerprint=(await t.one('SELECT md5($1) fingerprint',[JSON.stringify([p.action,m,p.rows])])).fingerprint;
  20. 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;}
  21. await t.any("SELECT pg_advisory_xact_lock(hashtext('xs-credit:'||$1))",[co]);
  22. const out={modelId:m,createdCommon:0,createdAddon:0,updated:0,duplicates:0,rewardEvidence:0,conflicts:[],postimages:[]};
  23. 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)]);}
  24. for(const incoming of p.rows){
  25. const gid=Number(incoming.generalId);if(!Number.isSafeInteger(gid)||gid<1)throw Error('Invalid incoming ID');
  26. 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;
  27. if(p.action==='append'){
  28. if(cc&&(cc.teachingManaged||Number(cc.status)===-2||!legacy(cc))){out.conflicts.push({generalId:gid,reason:'新系统课次已受保护'});continue;}
  29. let item=cc?Number(cc.itemId):Number(incoming.addon?.id||gid);
  30. 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;
  31. if(cc&&aa)continue;
  32. const sourceKey='legacy:model:'+m+':general:'+gid;
  33. 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;}}
  34. 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]);
  35. if(orphan){out.conflicts.push({generalId:gid,reason:'已有同源孤立附表,保留待核对'});continue;}
  36. 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};
  37. if(!common.objectId||!addon.objectId)throw Error('Missing object IDs');
  38. if(m===60){
  39. 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;
  40. const duplicate=key&&await t.oneOrNone('SELECT "objectId" FROM "MemoryPracticeRecord" WHERE company=$1 AND "rewardTaskKey"=$2 LIMIT 1',[co,key]);
  41. addon.rewardTaskKey=duplicate?null:key;addon.rewardDuplicate=Boolean(duplicate);if(duplicate)out.duplicates++;
  42. // 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.
  43. await t.none("SET LOCAL session_replication_role='replica'");
  44. await insert(s.className,addon,s.addonColumns);await t.none("SET LOCAL session_replication_role='origin'");
  45. 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++;}
  46. }else await insert(s.className,addon,s.addonColumns);
  47. out.createdAddon++;
  48. if(!cc){await insert('CommonModel',common,s.commonColumns);out.createdCommon++;}
  49. }else{
  50. if(!cc){out.conflicts.push({generalId:gid,reason:'主记录不存在'});continue;}
  51. 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;
  52. 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))){
  53. out.conflicts.push({generalId:gid,reason:'新系统已有修改、版本变化或来源不可确认'});continue;
  54. }
  55. 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;}
  56. const fields=Object.keys(incoming.values||{});if(!fields.length||fields.some(f=>!s.fields.includes(f)))throw Error('Invalid patch fields');
  57. if(m===60&&(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;}
  58. if(m===60)await t.none("SET LOCAL session_replication_role='replica'");
  59. 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)]);
  60. 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]);}
  61. out.updated++;
  62. }
  63. const after=await t.oneOrNone(readSql,[co,m,[gid]]);if(!after?.common||!after.addon)throw Error('Atomic projection verification failed');
  64. 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');}
  65. if(m===60&&after.reward&&!(Number(after.reward.amount)===0&&['historical','exception'].includes(after.reward.status)))throw Error('Historical reward verification failed');
  66. await directBaseline(t,co,m,gid,after);
  67. out.postimages.push(after);
  68. }
  69. if(!/^[A-Za-z0-9:._-]{1,160}$/.test(p.batchKey||''))throw Error('Invalid batch key');
  70. 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)]);
  71. return out;
  72. });
  73. return response.json({success:true,data:result});
  74. }catch(e){return response.status(500).json({success:false,message:String(e.message||e)});}
  75. }