export function readPressurePolicy(kind,params){ const operation=String(kind==='app'?params.action:params.operation||''); const operations=['ops/users/list','ops/users/detail','ops/stores/list','ops/stores/detail','ops/coaches/list','ops/coaches/detail','ops/schedule/calendar','ops/payroll/overview','ops/payroll/preview','ops/payroll/detail','ops/payroll/export','ops/dashboard/summary','ops/dashboard/schedule-trend','ops/lessons/list','ops/lessons/detail','ops/learning-reports/list','ops/learning-reports/detail','ops/learning-reports/export','ops/vocabulary/list','ops/vocabulary/tree','ops/special-training/list']; const actions=['app_review_page','app_companion_overview','app_learning_overview','app_account_overview','app_lesson_study','app_lesson_report','order_list','order_get','node_list','content_list','content_get','e_record_list','e_record_detail','app_exam_history']; return{heavy:(kind==='ops'?operations:kind==='app'?actions:[]).includes(operation),cache:kind==='ops'&&operation==='ops/lessons/list',operation}; } export function readPressureStable(value){ if(Array.isArray(value))return '['+value.map(readPressureStable).join(',')+']'; if(value&&typeof value==='object')return '{'+Object.keys(value).filter(k=>value[k]!==undefined).sort().map(k=>JSON.stringify(k)+':'+readPressureStable(value[k])).join(',')+'}'; return JSON.stringify(value)??'null'; } export function readPressureDb(database,deadline,inTransaction=false){ const methods=new Set(['query','one','oneOrNone','none','any','many','manyOrNone','result','multi','multiResult']); const check=()=>{const remaining=deadline-Date.now();if(remaining<=0)throw Object.assign(Error('查询暂未完成,请稍后重试'),{status:408});return remaining;}; return new Proxy(Object.create(null),{get(_target,key){ const target=database; if(key==='__xiaoshuReadBudget')return true; const value=target[key]; if(key==='tx'&&typeof value==='function')return(...args)=>{const callback=args.at(-1);if(typeof callback!=='function')throw Error('Invalid transaction callback');args[args.length-1]=async tx=>{await tx.none("SET LOCAL statement_timeout='"+Math.max(1,check())+"ms'");return callback(readPressureDb(tx,deadline,true));};return value.apply(target,args);}; if(methods.has(key)&&typeof value==='function')return(...args)=>{ const pgp=target.$config?.pgp; if(!inTransaction&&typeof args[0]==='string'&&typeof target.task==='function'&&typeof pgp?.as?.format==='function'&&['query','one','oneOrNone','none','any','many','manyOrNone','result'].includes(key))return target.task(async connection=>{ const sql=pgp.as.format(args[0],args[1]),timeout=Math.max(1,Math.min(12000,check())); let result; try{const results=await connection.multiResult("BEGIN;SET LOCAL statement_timeout='"+timeout+"ms';"+sql+"\n;COMMIT;");result=results.at(-2);} catch(error){await connection.none('ROLLBACK').catch(()=>{});throw error;} const rows=result.rows,errors=pgp.errors; if((key==='one'||key==='many')&&!rows.length)throw new errors.QueryResultError(errors.queryResultErrorCode.noData,result,args[0],args[1]); if((key==='one'||key==='oneOrNone')&&rows.length>1)throw new errors.QueryResultError(errors.queryResultErrorCode.multiple,result,args[0],args[1]); if(key==='none'&&rows.length)throw new errors.QueryResultError(errors.queryResultErrorCode.notEmpty,result,args[0],args[1]); const output=key==='result'?result:key==='none'?null:key==='one'||key==='oneOrNone'?rows[0]||null:rows; return typeof args[2]==='function'?args[2](output):output; }); const execute=async tx=>{await tx.none("SET LOCAL statement_timeout='"+Math.max(1,check())+"ms'");check();return tx[key](...args);}; return inTransaction?execute(target):target.tx(execute); }; return typeof value==='function'?value.bind(target):value; }}); } export async function dispatchReadPressure(database,request,response,kind,execute,authorize){ const params=request.body?.params||{},policy=readPressurePolicy(kind,params); if(!policy.heavy)return execute(database); const availabilitySymbol=Symbol.for('xiaoshu.read-pressure.availability.v1'); const availability=globalThis[availabilitySymbol]||(globalThis[availabilitySymbol]=new WeakMap()),availabilityKey=database.$pool||database; let available=availability.get(availabilityKey); if(!available||available.untilnew Promise(resolve=>setTimeout(resolve,ms)); const start=Date.now(),deadline=start+14_000; let cacheKey,revision,identity,ownsCache=false,permit; const refresh=params.refresh===true; const refreshSince=refresh?(await database.one('SELECT clock_timestamp()::text stamp')).stamp:null; const busy=()=>{throw Object.assign(Error('查询较多,请稍后重试'),{status:429,retryAfterMs:2000});}; const cacheRead=async()=>{const row=(await database.query('SELECT value,expires_at,GREATEST(0,extract(epoch FROM expires_at-clock_timestamp())*1000)::int remaining_ms FROM xs_read_cache WHERE key=$1 AND revision=$2 AND expires_at>clock_timestamp() AND value IS NOT NULL',[cacheKey,revision]))[0];return row;}; try{ if(policy.cache){ identity=await authorize();if(!identity?.company||!identity?.scope)throw Error('Missing read scope'); revision=(await database.one("SELECT xs_read_version($1,'lessons') revision",[identity.company])).revision; const {refresh:_refresh,requestId:_requestId,...keyParams}=params; cacheKey=crypto.createHash('sha256').update(readPressureStable([identity.company,identity.scope,policy.operation,keyParams])).digest('hex'); if(!refresh){const value=await cacheRead();if(value)return response.json({...value.value,readCache:{cached:true,remainingMs:value.remaining_ms}});} while(Date.now()excluded.revision OR xs_read_cache.expires_at<=clock_timestamp() OR ($6 AND xs_read_cache.expires_at<$7::timestamptz+interval '30 seconds')) RETURNING key",[cacheKey,identity.company,identity.scope,revision,owner,refresh,refreshSince]); if(claimed.length){ownsCache=true;break;} // A concurrent force refresh shares only the result produced after this call started. const result=(await database.query("SELECT value,expires_at,GREATEST(0,extract(epoch FROM expires_at-clock_timestamp())*1000)::int remaining_ms,expires_at>=$3::timestamptz+interval '30 seconds' refreshed FROM xs_read_cache WHERE key=$1 AND revision=$2 AND value IS NOT NULL AND expires_at>clock_timestamp()",[cacheKey,revision,refreshSince]))[0]; if(result&&(!refresh||result.refreshed===true))return response.json({...result.value,readCache:{cached:true,remainingMs:result.remaining_ms}}); await wait(100+Math.floor(Math.random()*80)); } if(!ownsCache)busy(); } const permitDeadline=Math.min(deadline,Date.now()+2000); while(Date.now(){}); if(ownsCache)await database.none('UPDATE xs_read_cache SET owner=NULL,lease_until=NULL WHERE key=$1 AND owner=$2',[cacheKey,owner]).catch(()=>{}); } } export function readPressureRuntimeSource(){return [readPressurePolicy,readPressureStable,readPressureDb,dispatchReadPressure].map(fn=>fn.toString()).join('\n');}