import { EventEmitter } from "events" ; import { callback } from "awaiting" ; import { callback2 } from "@cocalc/util/async-utils" ; import { close , len } from "@cocalc/util/misc" ; import { PostgreSQL , QueryOptions , QueryResult } from "./types" ; import { ChangeEvent , Changes } from "./changefeed" ; const { all_results } = require ( "../postgres-base" ) ; type SetOfAccounts = { [ account_id ] : boolean ; } ; type SetOfProjects = { [ project_id ] : boolean ; } ; type State = "init" | "ready" | "closed" ; export class ProjectAndUserTracker extends EventEmitter { private state : State = "init" ; private db : PostgreSQL ; private feed : Changes ; private accounts : SetOfAccounts = { } ; private users : { [ project_id ] : SetOfAccounts ; } = { } ; private projects : { [ account_id ] : SetOfProjects ; } = { } ; private collabs : { [ account_id ] : { [ account_id ] : number ; } ; } = { } ; private register_todo : { [ account_id ] : Function [ ] ; } = { } ; private do_register_lock : boolean = false ; constructor ( db ) { super ( ) ; this . db = db ; } private assert_state ( state , f ) { if ( this . state != state ) { throw Error ( `${ f }: state must be ${ state } but it is ${ this . state }` ) ; } } async init ( ) { this . assert_state ( "init" , "init" ) ; const dbg = this . dbg ( "init" ) ; dbg ( "Initializing Project and user tracker..." ) ; this . setMaxListeners ( 1000 ) ; try { this . feed = await callback2 ( this . db . changefeed , { table : "projects" , select : { project_id : "UUID" } , watch : [ "users" ] , where : { } , } ) ; dbg ( "Success" ) ; } catch ( err ) { this . handle_error ( err ) ; return ; } this . feed . on ( "change" , this . handle_change . bind ( this ) ) ; this . feed . on ( "error" , this . handle_error . bind ( this ) ) ; this . feed . on ( "close" , ( ) => this . handle_error ( "changefeed closed" ) ) ; this . set_state ( "ready" ) ; } private dbg ( f ) { return this . db . _dbg ( `Tracker.${ f }` ) ; } private handle_error ( err ) { if ( this . state == "closed" ) return ; const dbg = this . dbg ( "handle_error" ) ; dbg ( `err='${ err }'` ) ; this . emit ( "error" , err ) ; this . close ( ) ; } private set_state ( state ) { this . state = state ; this . emit ( state ) ; } close ( ) { if ( this . state == "closed" ) { return ; } this . set_state ( "closed" ) ; this . removeAllListeners ( ) ; if ( this . feed != null ) { this . feed . close ( ) ; } if ( this . register_todo != null ) { for ( const account_id in this . register_todo ) { const callbacks = this . register_todo [ account_id ] ; if ( callbacks != null ) { for ( const cb of callbacks ) { cb ( "closed" ) ; } } } } close ( this ) ; this . state = "closed" ; } private handle_change_delete ( old_val ) { this . assert_state ( "ready" , "handle_change_delete" ) ; const { project_id } = old_val ; if ( this . users [ project_id ] == null ) { return ; } for ( const account_id in this . users [ project_id ] ) { this . remove_user_from_project ( account_id , project_id ) ; } return ; } private handle_change ( x ) { this . assert_state ( "ready" , "handle_change" ) ; if ( x . action === "delete" ) { if ( x . old_val == null ) return ; this . handle_change_delete ( x . old_val ) ; } else { if ( x . new_val == null ) return ; this . handle_change_update ( x . new_val ) ; } } private async handle_change_update ( new_val ) { this . assert_state ( "ready" , "handle_change_update" ) ; const dbg = this . dbg ( "handle_change_update" ) ; dbg ( new_val ) ; const { project_id } = new_val ; let users ; try { users = await query ( this . db , { query : "SELECT jsonb_object_keys(users) AS account_id FROM projects" , where : { "project_id = $::UUID" : project_id } , } ) ; } catch ( err ) { this . handle_error ( err ) ; return ; } if ( this . users [ project_id ] == null ) { let any = false ; for ( const { account_id } of users ) { if ( this . accounts [ account_id ] ) { any = true ; break ; } } if ( ! any ) { return ; } } const users_now = { } ; for ( const { account_id } of users ) { users_now [ account_id ] = true ; } const users_before = this . users [ project_id ] != null ? this . users [ project_id ] : { } ; for ( const account_id in users_now ) { if ( ! users_before [ account_id ] ) { this . add_user_to_project ( account_id , project_id ) ; } } for ( const account_id in users_before ) { if ( ! users_now [ account_id ] ) { this . remove_user_from_project ( account_id , project_id ) ; } } } private add_user_to_project ( account_id , project_id ) { this . assert_state ( "ready" , "add_user_to_project" ) ; if ( this . projects [ account_id ] != null && this . projects [ account_id ] [ project_id ] ) { return ; } this . emit ( `add_user_to_project-${ account_id }` , project_id ) ; if ( this . users [ project_id ] == null ) { this . users [ project_id ] = { } ; } const users = this . users [ project_id ] ; users [ account_id ] = true ; if ( this . projects [ account_id ] == null ) { this . projects [ account_id ] = { } ; } const projects = this . projects [ account_id ] ; projects [ project_id ] = true ; if ( this . collabs [ account_id ] == null ) { this . collabs [ account_id ] = { } ; } const collabs = this . collabs [ account_id ] ; for ( const other_account_id in users ) { if ( collabs [ other_account_id ] != null ) { collabs [ other_account_id ] += 1 ; } else { collabs [ other_account_id ] = 1 ; this . emit ( `add_collaborator-${ account_id }` , other_account_id ) ; } const other_collabs = this . collabs [ other_account_id ] ; if ( other_collabs [ account_id ] != null ) { other_collabs [ account_id ] += 1 ; } else { other_collabs [ account_id ] = 1 ; this . emit ( `add_collaborator-${ other_account_id }` , account_id ) ; } } } private remove_user_from_project ( account_id , project_id , no_emit = false ) { this . assert_state ( "ready" , "remove_user_from_project" ) ; if ( ( account_id != null ? account_id . length : undefined ) !== 36 || ( project_id != null ? project_id . length : undefined ) !== 36 ) { throw Error ( "invalid account_id or project_id" ) ; } if ( ! ( this . projects [ account_id ] != null ? this . projects [ account_id ] [ project_id ] : undefined ) ) { return ; } if ( ! no_emit ) { this . emit ( `remove_user_from_project-${ account_id }` , project_id ) ; } if ( this . collabs [ account_id ] == null ) { this . collabs [ account_id ] = { } ; } for ( const other_account_id in this . users [ project_id ] ) { this . collabs [ account_id ] [ other_account_id ] -= 1 ; if ( this . collabs [ account_id ] [ other_account_id ] === 0 ) { delete this . collabs [ account_id ] [ other_account_id ] ; if ( ! no_emit ) { this . emit ( `remove_collaborator-${ account_id }` , other_account_id ) ; } } this . collabs [ other_account_id ] [ account_id ] -= 1 ; if ( this . collabs [ other_account_id ] [ account_id ] === 0 ) { delete this . collabs [ other_account_id ] [ account_id ] ; if ( ! no_emit ) { this . emit ( `remove_collaborator-${ other_account_id }` , account_id ) ; } } } delete this . users [ project_id ] [ account_id ] ; delete this . projects [ account_id ] [ project_id ] ; } public async register ( account_id ) { await callback ( this . register_cb . bind ( this ) , account_id ) ; } private register_cb ( account_id , cb ) { const dbg = this . dbg ( `register(account_id="${ account_id }"` ) ; if ( this . accounts [ account_id ] != null ) { dbg ( `already registered -- listener counts ${ JSON . stringify ( this . listener_counts ( account_id ) ) }` ) ; cb ( ) ; return ; } if ( len ( this . register_todo ) === 0 ) { this . register_todo [ account_id ] = [ cb ] ; this . do_register ( ) ; } else { const v = this . register_todo [ account_id ] ; if ( v != null ) { v . push ( cb ) ; } else { this . register_todo [ account_id ] = [ cb ] ; } } } private async do_register ( ) { if ( this . state != "ready" ) return ; let account_id = undefined ; for ( account_id in this . register_todo ) break ; if ( account_id == null ) return ; const dbg = this . dbg ( `do_register(account_id="${ account_id }")` ) ; dbg ( "registering account" ) ; if ( this . do_register_lock ) throw Error ( "do_register MUST NOT be called twice at once!" ) ; this . do_register_lock = true ; try { let projects ; try { projects = await query ( this . db , { query : "SELECT project_id, json_agg(o) as users FROM (SELECT project_id, jsonb_object_keys(users) AS o FROM projects WHERE users ? $1::TEXT ORDER BY last_edited DESC LIMIT 10000) s group by s.project_id" , params : [ account_id ] , } ) ; } catch ( err ) { const e = `error registering '${ account_id }' -- err=${ err }` ; dbg ( e ) ; this . handle_error ( e ) ; return ; } this . accounts [ account_id ] = true ; dbg ( "now adding all users to project tracker -- start" ) ; for ( const project of projects ) { if ( this . users [ project . project_id ] != null ) { continue ; } else { for ( const collab_account_id of project . users ) { if ( collab_account_id == null ) { continue ; } this . add_user_to_project ( collab_account_id , project . project_id ) ; } } } dbg ( "successfully registered -- stop" ) ; const callbacks = this . register_todo [ account_id ] ; if ( callbacks != null ) { for ( const cb of callbacks ) { cb ( ) ; } delete this . register_todo [ account_id ] ; } } finally { this . do_register_lock = false ; } if ( len ( this . register_todo ) > 0 ) { this . do_register ( ) ; } } public unregister ( account_id ) { if ( ! this . accounts [ account_id ] ) return ; const v = [ ] ; for ( const project_id in this . projects [ account_id ] ) { v . push ( project_id ) ; } delete this . accounts [ account_id ] ; for ( const project_id of v ) { let need = false ; for ( const other_account_id in this . users [ project_id ] ) { if ( this . accounts [ other_account_id ] != null ) { need = true ; break ; } } if ( ! need ) { for ( const other_account_id in this . users [ project_id ] ) { this . remove_user_from_project ( other_account_id , project_id , true ) ; } delete this . users [ project_id ] ; } } } public get_projects ( account_id ) { if ( ! this . accounts [ account_id ] ) { this . register ( account_id ) ; return { } ; } return this . projects [ account_id ] != null ? this . projects [ account_id ] : { } ; } public get_collabs ( account_id ) { if ( this . state == "closed" ) return { } ; return this . collabs [ account_id ] != null ? this . collabs [ account_id ] : { } ; } private listener_counts ( account_id ) { const x = { } ; for ( const e of [ "add_user_to_project" , "remove_user_from_project" , "add_collaborator" , "remove_collaborator" , ] ) { const event = e + "-" + account_id ; x [ event ] = this . listenerCount ( event ) ; } return x ; } } function all_query ( db , opts , cb ) { if ( opts == null ) { throw Error ( "opts must not be null" ) ; } opts . cb = all_results ( cb ) ; db . _query ( opts ) ; } async function query ( db , opts ) { return await callback ( all_query , db , opts ) ; }