xiaoshu-app/scripts/lib/read-pressure-runtime.mjs

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.until<Date.now()){available={...await database.one("SELECT to_regclass('xs_read_permit') IS NOT NULL installed"),until:Date.now()+10000};availability.set(availabilityKey,available);}
 if(!available.installed)return execute(database);
 const crypto=await import('node:crypto'),owner=crypto.randomUUID(),wait=ms=>new 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()<deadline){
    const claimed=await database.query("INSERT INTO xs_read_cache(key,company,scope,revision,owner,lease_until) VALUES($1,$2,$3,$4,$5,clock_timestamp()+interval '16 seconds') ON CONFLICT(key) DO UPDATE SET revision=excluded.revision,owner=excluded.owner,lease_until=excluded.lease_until,value=NULL,expires_at='epoch' WHERE (xs_read_cache.owner IS NULL OR xs_read_cache.lease_until<=clock_timestamp()) AND (xs_read_cache.revision<>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()<permitDeadline){
   const claimed=await database.query("UPDATE xs_read_permit SET owner=$1,lease_until=clock_timestamp()+interval '16 seconds' WHERE slot=(SELECT slot FROM xs_read_permit WHERE slot<=(SELECT concurrency FROM xs_read_pressure_config WHERE id) AND (owner IS NULL OR lease_until<=clock_timestamp()) ORDER BY slot FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING slot",[owner]);
   if(claimed.length){permit=claimed[0].slot;break;}await wait(80+Math.floor(Math.random()*60));
  }
  if(!permit)busy();
  const bounded=readPressureDb(database,Math.min(deadline,Date.now()+12000));
  let envelope;const original=response.json;
  if(ownsCache)response.json=function(body){envelope=body?.success===true?{...body,readCache:{cached:false,remainingMs:30000}}:body;return original.call(this,envelope);};
  try{return await execute(bounded);}finally{
   response.json=original;
   if(ownsCache&&envelope?.success===true)await database.none("UPDATE xs_read_cache SET value=$3::jsonb,expires_at=clock_timestamp()+interval '30 seconds',owner=NULL,lease_until=NULL WHERE key=$1 AND owner=$2",[cacheKey,owner,JSON.stringify(envelope)]);
  }
 }finally{
  if(permit)await database.none('UPDATE xs_read_permit SET owner=NULL,lease_until=NULL WHERE slot=$1 AND owner=$2',[permit,owner]).catch(()=>{});
  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');}