const THROTTLE = true ; let THROTTLE_MS = process . env . SMC_TEST ? 10 : 500 ; if ( process . env . POSTGRES_THROTTLE_CHANGEFEED_MS != null ) { THROTTLE_MS = parseInt ( process . env . POSTGRES_THROTTLE_CHANGEFEED_MS ) ; } import { EventEmitter } from "events" ; import { callback } from "awaiting" ; import { once } from "@cocalc/util/async-utils" ; import { close , copy } from "@cocalc/util/misc" ; const { one_result , all_results } = require ( "../postgres-base" ) ; import { PostgreSQL , QueryWhere } from "./types" ; interface QueryOpts { db : PostgreSQL ; select : string [ ] ; table : string ; where : QueryWhere ; one : boolean ; } interface TableQuery { select : string [ ] ; where : QueryWhere ; } function key ( obj ) { return `query-${ JSON . stringify ( obj ) }` ; } type State = "ready" | "closed" ; class ThrottledTableQueue extends EventEmitter { private table : string ; private queue : { [ key ] : TableQuery ; } = { } ; private db : PostgreSQL ; private process_timer : any ; private interval_ms : number ; private state : State = "ready" ; constructor ( db , table , interval_ms ) { super ( ) ; this . db = db ; this . table = table ; this . interval_ms = interval_ms ; this . setMaxListeners ( 100 ) ; } private dbg ( f ) { return this . db . _dbg ( `ThrottledTableQueue('${ this . table }').${ f }` ) ; } public close ( ) { if ( this . state == "closed" ) { return ; } if ( this . process_timer != null ) { clearTimeout ( this . process_timer ) ; } for ( const k in this . queue ) { this . emit ( k , "closed" ) ; } this . emit ( "closed" ) ; this . removeAllListeners ( ) ; close ( this ) ; this . state = "closed" ; } public enqueue ( query ) { if ( this . state == "closed" ) { throw Error ( "trying to enqueue after close" ) ; } const k = key ( query ) ; this . queue [ k ] = query ; if ( this . process_timer == null ) { this . dbg ( "enqueue" ) ( `will process queue in ${ this . interval_ms }ms...` ) ; this . process_timer = setTimeout ( this . process_queue . bind ( this ) , this . interval_ms ) ; } return k ; } private async process_queue ( ) { const dbg = this . dbg ( "process_queue" ) ; delete this . process_timer ; const queue = copy ( this . queue ) ; this . queue = { } ; for ( const k in queue ) { dbg ( k ) ; const { select , where } = queue [ k ] ; try { const result = await callback ( one_query , this . db , select , this . table , where ) ; if ( this . state == "closed" ) return ; dbg ( "success" , k ) ; this . emit ( k , undefined , result ) ; } catch ( err ) { if ( this . state == "closed" ) return ; dbg ( "fail" , k ) ; this . emit ( k , err , undefined ) ; } } } } const throttled_table_queues = { } ; function throttled_table_queue ( db , table , interval_ms ) { if ( throttled_table_queues [ table ] != null ) { return throttled_table_queues [ table ] ; } return ( throttled_table_queues [ table ] = new ThrottledTableQueue ( db , table , interval_ms ) ) ; } export async function query ( opts ) { if ( THROTTLE && opts . one ) { const Table = throttled_table_queue ( opts . db , opts . table , THROTTLE_MS ) ; const k = Table . enqueue ( { select : opts . select , where : opts . where } ) ; const [ err , result ] = await once ( Table , k ) ; if ( err != null ) { throw err ; } return result ; } return await callback ( opts . one ? one_query : all_query , opts . db , opts . select , opts . table , opts . where ) ; } function all_query ( db , select , table , where , cb ) { db . _query ( { select , table , where , cb : all_results ( cb ) } ) ; } function one_query ( db , select , table , where , cb ) { db . _query ( { select , table , where , cb : one_result ( cb ) } ) ; }