import debug from "debug" ; const L = debug ( "project:sync:usage-info" ) ; import { once } from "@cocalc/util/async-utils" ; import { SyncTable , SyncTableState } from "@cocalc/sync/table" ; import { close , merge } from "@cocalc/util/misc" ; import { UsageInfoServer } from "../usage-info" ; import { UsageInfo , ImmutableUsageInfo } from "../usage-info/types" ; class UsageInfoTable { private readonly table ? : SyncTable ; private readonly project_id : string ; private readonly servers : { [ path ] : UsageInfoServer ; } = { } ; private readonly log : Function ; constructor ( table , project_id ) { this . project_id = project_id ; this . log = L . extend ( "table" ) ; this . table = table ; this . setup_watchers ( ) ; } public close ( ) { this . log ( "close" ) ; for ( const path in this . servers ) { this . stop_server ( path ) ; } close ( this ) ; } private async setup_watchers ( ) { if ( this . table == null ) return ; if ( this . table . get_state ( ) == ( "init" as SyncTableState ) ) { await once ( this . table , "state" ) ; } if ( this . table . get_state ( ) != ( "connected" as SyncTableState ) ) { return ; } this . table . get ( ) ?. forEach ( ( val ) => { const path = val . get ( "path" ) ; if ( path == null ) return ; if ( this . servers [ path ] == null ) return ; } ) ; this . log ( "setting up 'on.change'" ) ; this . table . on ( "change" , this . handle_change_event . bind ( this ) ) ; } private async remove_stale_servers ( ) { if ( this . table == null ) return ; if ( this . table . get_state ( ) != ( "connected" as SyncTableState ) ) return ; const paths = [ ] ; this . table . get ( ) ?. forEach ( ( val ) => { const path = val . get ( "path" ) ; if ( path == null ) return ; paths . push ( path ) ; } ) ; for ( const path of Object . keys ( this . servers ) ) { if ( ! paths . includes ( path ) ) { this . stop_server ( path ) ; } } } private is_ready ( ) { return ! ! this . table ?. is_ready ( ) ; } private get_table ( ) { if ( ! this . is_ready ( ) || this . table == null ) { throw Error ( "table not ready" ) ; } return this . table ; } async set ( obj ) { this . get_table ( ) . set ( merge ( { project_id : this . project_id } , obj ) , "shallow" ) ; await this . get_table ( ) . save ( ) ; } public get ( path ) { const x = this . get_table ( ) . get ( JSON . stringify ( [ this . project_id , path ] ) ) ; if ( x == null ) return x ; return ( x as unknown ) as ImmutableUsageInfo ; } private handle_change_event ( keys ) { for ( const key of keys ) { this . handle_change ( JSON . parse ( key ) [ 1 ] ) ; } this . remove_stale_servers ( ) ; } private handle_change ( path ) { this . log ( "handle_change" , path ) ; const cur = this . get ( path ) ; if ( cur == null ) return ; this . ensure_watching ( path ) ; this . set ( { path } ) ; } private ensure_watching ( path ) { if ( this . servers [ path ] != null ) { return ; } try { this . start_watching ( path ) ; } catch ( err ) { this . log ( "failed to start watching" , err ) ; } } private start_watching ( path ) { this . log ( `start_watching ${ path }` ) ; if ( this . servers [ path ] != null ) return ; const server = new UsageInfoServer ( path ) ; server . on ( "usage" , ( usage ) => { try { if ( ! this . is_ready ( ) ) return ; this . set ( { path , usage } ) ; } catch ( err ) { this . log ( `compute_listing("${ path }") error: "${ err }"` ) ; } } ) ; server . start ( ) ; this . servers [ path ] = server ; } private stop_server ( path ) { const s = this . servers [ path ] ; if ( s == null ) return ; delete this . servers [ path ] ; s . stop ( ) ; this . remove_path ( path ) ; } private async remove_path ( path ) { if ( ! this . is_ready ( ) ) return ; this . log ( "remove_path" , path ) ; await this . get_table ( ) . delete ( { project_id : this . project_id , path } ) ; } } let usage_info_table = undefined ; export function register_usage_info_table ( table , project_id ) { L ( "register_usage_info_table" ) ; if ( usage_info_table != null ) { usage_info_table . close ( ) ; } usage_info_table = new UsageInfoTable ( table , project_id ) ; } export function get_usage_info_table ( ) { return usage_info_table ; }