|
|
@@ -0,0 +1,90 @@
|
|
|
+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');}
|
彭峰