// 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=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_untilnow() 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$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{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'};});}