xiaoshu-app/scripts/cloud/lesson-export.js

// Persistent jobs are processed by scripts/lesson-export-worker.mjs, never by detached request work.
function lessonExportProjection(job){return{jobId:job.id,status:job.status,total:number(job.total),processed:number(job.processed),error:job.error||'',snapshotReady:job.snapshot_ready===true,expiresAt:job.expires_at};}
function lessonExportScope(context){return JSON.stringify([companyIdOf(context,{}),context.storeId||0,roleOf(context)]);}
function lessonExportFilters(input){const out={};for(const key of ['studentId','search','coachName','storeName','courseName','dateFrom','dateTo'])if(input[key]!==undefined)out[key]=input[key];return out;}
async function lessonExportCreate(context,input){
 requireRole(context,['ops-manager','ops-staff','ops-auditor']);const company=companyIdOf(context,input),actor=text(context.current.id),key=text(input.idempotencyKey),filters=lessonExportFilters(input),scope=lessonExportScope(context);
 if(!company||key.length<16||key.length>100)fail(400,'导出请求编号无效');
 return Psql.tx(async db=>{
  await db.one('SELECT pg_advisory_xact_lock(hashtext($1),hashtext($2))',[company,actor]);
  await db.none('DELETE FROM xs_history_export_job WHERE company=$1 AND actor=$2 AND expires_at<=now()',[company,actor]);
  const prior=(await db.query('SELECT id,filters,scope,status,total,processed,error,snapshot_ready,expires_at FROM xs_history_export_job WHERE company=$1 AND actor=$2 AND request_key=$3 AND expires_at>now()',[company,actor,key]))[0];
  if(prior){if(JSON.stringify(prior.filters)!==JSON.stringify(JSON.parse(JSON.stringify(filters)))||prior.scope!==scope){
    // PostgreSQL jsonb canonicalizes key order; compare its canonical representation instead.
    const same=await db.one('SELECT filters=$2::jsonb same FROM xs_history_export_job WHERE id=$1',[prior.id,JSON.stringify(filters)]);if(!same.same||prior.scope!==scope)fail(409,'导出请求条件已变化');
   }return lessonExportProjection(prior);}
  const active=(await db.query("SELECT id,filters,scope,status,total,processed,error,snapshot_ready,expires_at FROM xs_history_export_job WHERE company=$1 AND actor=$2 AND status IN('queued','running') AND expires_at>now()",[company,actor]))[0];
  if(active){const same=await db.one('SELECT filters=$2::jsonb same FROM xs_history_export_job WHERE id=$1',[active.id,JSON.stringify(filters)]);if(!same.same||active.scope!==scope)fail(409,'已有导出进行中，请等待或取消');return lessonExportProjection(active);}
  const job=await db.one('INSERT INTO xs_history_export_job(company,actor,scope,request_key,filters) VALUES($1,$2,$3,$4,$5::jsonb) RETURNING *',[company,actor,scope,key,JSON.stringify(filters)]);return lessonExportProjection(job);
 });
}
async function lessonExportOwned(context,input,db=Psql,includeFile=false){
 requireRole(context,['ops-manager','ops-staff','ops-auditor']);
 const job=(await db.query('SELECT id,status,total,processed,error,snapshot_ready,expires_at'+(includeFile?',file_base64':'')+' FROM xs_history_export_job WHERE id=$1 AND company=$2 AND actor=$3 AND scope=$4 AND expires_at>now()',[text(input.jobId),companyIdOf(context,input),text(context.current.id),lessonExportScope(context)]))[0];
 if(!job)fail(404,'未找到有效导出任务');return job;
}
async function lessonExportStatus(context,input){return lessonExportProjection(await lessonExportOwned(context,input));}
async function lessonExportCancel(context,input){const job=await lessonExportOwned(context,input);await Psql.none("UPDATE xs_history_export_job SET status='cancelled',lease=NULL,lease_until=NULL,file_base64=NULL WHERE id=$1 AND status IN('queued','running')",[job.id]);return lessonExportStatus(context,input);}
async function lessonExportDownload(context,input){const job=await lessonExportOwned(context,input,Psql,true);if(job.status!=='ready'||!job.file_base64)fail(409,'导出尚未完成');return{fileName:'训练记录.xlsx',contentType:'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',base64:job.file_base64};}
async function lessonExportClaim(context){
 requireSuperAdmin(context);
 return Psql.tx(async db=>{
  await db.one('SELECT pg_advisory_xact_lock(1736201,1)');
  await db.none('DELETE FROM xs_history_export_job WHERE expires_at<=now()');
  await db.none("UPDATE xs_history_export_job SET status='failed',error='后台任务多次中断，请重新导出',lease=NULL,lease_until=NULL WHERE status='running' AND lease_until<now() AND attempts>=3");
  const active=await db.one("SELECT EXISTS(SELECT 1 FROM xs_history_export_job WHERE status='running' AND lease_until>now()) active");if(active.active)return{};
  const job=(await db.query("SELECT * FROM xs_history_export_job WHERE status='queued' OR (status='running' AND lease_until<now() AND attempts<3) ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED"))[0];if(!job)return{};
  const claimed=await db.one("UPDATE xs_history_export_job SET status='running',processed=0,error='',lease=md5(random()::text||clock_timestamp()::text),lease_until=now()+interval '2 minutes',attempts=attempts+1 WHERE id=$1 RETURNING *",[job.id]);
  return{...lessonExportProjection(claimed),lease:claimed.lease,snapshotReady:claimed.snapshot_ready};
 });
}
async function lessonExportLeased(context,input,db=Psql){
 requireSuperAdmin(context);const job=(await db.query("SELECT * FROM xs_history_export_job WHERE id=$1 AND lease=$2 AND status='running' AND lease_until>now() AND expires_at>now() FOR UPDATE",[text(input.jobId),text(input.lease)]))[0];if(!job)fail(409,'导出租约已失效');return job;
}
async function lessonExportSnapshot(context,input){
 return Psql.tx(async db=>{
  await db.none("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ");await db.none("SET LOCAL statement_timeout='60s'");
  const job=await lessonExportLeased(context,input,db);if(job.snapshot_ready)return lessonExportProjection(job);
  const user=await new Parse.Query('_User').get(job.actor,{useMasterKey:true}),owner=await auth({user});
  if(lessonExportScope(owner)!==job.scope||companyIdOf(owner,{})!==job.company)fail(403,'导出账号权限已变化');
  requireRole(owner,['ops-manager','ops-staff','ops-auditor']);
  const plan=lessonHistoryPlan(owner,job.filters,null,null,true),jobParameter=plan.args.length+1;
  // Keep the large snapshot inside PostgreSQL. Only 500 rows enter JS at once.
  const inserted=await db.one(`WITH inserted AS (
   INSERT INTO xs_history_export_row(job_id,position,value)
   SELECT $${jobParameter}::text,element.ordinality::int,element.value FROM (${plan.q}) snapshot
   CROSS JOIN LATERAL jsonb_array_elements(snapshot.items) WITH ORDINALITY element(value,ordinality) RETURNING 1
  ) SELECT count(*)::int total FROM inserted`,[...plan.args,job.id]);
  let position=0;
  while(position<number(inserted.total)){
   const batch=await db.query('SELECT position,value FROM xs_history_export_row WHERE job_id=$1 AND position>$2 ORDER BY position LIMIT 500',[job.id,position]);
   if(!batch.length)fail(409,'导出快照不完整');
   const resolved=(await lessonHistoryTypeRows(batch.map(row=>row.value),db)).map(lessonHistoryRow);
   await db.none('UPDATE xs_history_export_row r SET value=b.value FROM jsonb_to_recordset($2::jsonb) b(position integer,value jsonb) WHERE r.job_id=$1 AND r.position=b.position',[job.id,JSON.stringify(batch.map((row,index)=>({position:row.position,value:resolved[index]})))]);
   position=batch.at(-1).position;
  }
  const updated=await db.one("UPDATE xs_history_export_job SET total=$2,snapshot_ready=true,lease_until=now()+interval '2 minutes' WHERE id=$1 RETURNING *",[job.id,number(inserted.total)]);return lessonExportProjection(updated);
 });
}
async function lessonExportChunk(context,input){
 return Psql.tx(async db=>{const job=await lessonExportLeased(context,input,db);if(!job.snapshot_ready)fail(409,'导出快照尚未完成');const offset=Math.max(0,Math.floor(number(input.offset)));
 const rows=await db.query('SELECT value FROM xs_history_export_row WHERE job_id=$1 AND position>$2 ORDER BY position LIMIT 500',[job.id,offset]);
 await db.none("UPDATE xs_history_export_job SET processed=$2,lease_until=now()+interval '2 minutes' WHERE id=$1",[job.id,Math.min(job.total,offset+rows.length)]);
 return{items:rows.map(r=>r.value),total:job.total,offset};});
}
async function lessonExportFinish(context,input){
 return Psql.tx(async db=>{const job=await lessonExportLeased(context,input,db);const data=text(input.base64);
 if(!job.snapshot_ready||number(input.rowCount)!==job.total||!data||data.length>40_000_000||!/^[A-Za-z0-9+/]+={0,2}$/.test(data))fail(400,'导出文件或记录数量无效');
 await db.none("UPDATE xs_history_export_job SET status='ready',processed=total,file_base64=$2,lease=NULL,lease_until=NULL WHERE id=$1",[job.id,data]);return{jobId:job.id,status:'ready'};});
}
async function lessonExportFail(context,input){return Psql.tx(async db=>{const job=await lessonExportLeased(context,input,db);await db.none("UPDATE xs_history_export_job SET status='failed',error='导出未能完成，请稍后重试',lease=NULL,lease_until=NULL WHERE id=$1",[job.id]);return{jobId:job.id,status:'failed'};});}
async function lessonExportMaintenance(context){
 requireSuperAdmin(context);const installed=await Psql.one("SELECT to_regprocedure('xs_prune_read_revisions()') IS NOT NULL installed");
 if(installed.installed)await Psql.one('SELECT xs_prune_read_revisions()');
 const cache=await Psql.one("SELECT to_regclass('xs_read_cache') IS NOT NULL installed");if(cache.installed)await Psql.none('DELETE FROM xs_read_cache WHERE expires_at<now() AND (lease_until IS NULL OR lease_until<now())');return{completed:true};
}

async function lessonExportRelease(context,input){return Psql.tx(async db=>{const job=await lessonExportLeased(context,input,db);await db.none("UPDATE xs_history_export_job SET lease_until=now()-interval '1 second' WHERE id=$1",[job.id]);return{jobId:job.id,status:'running'};});}