const CLOSE_DELAY_MS = 5 * 60 * 1000 ; const MAX_CONNECTIONS = 500 ; const MAX_CONNECTIONS_FROM_ONE_CLIENT = 10 ; import { synctable_no_changefeed , synctable_no_database , SyncTable , VersionedChange , set_debug , } from "@cocalc/sync/table" ; const _ = set_debug ; import { init_syncdoc } from "./sync-doc" ; import { key , register_synctable } from "./open-synctables" ; import { reuseInFlight } from "async-await-utils/hof" ; import { once } from "@cocalc/util/async-utils" ; import { delay } from "awaiting" ; import { close , deep_copy , len } from "@cocalc/util/misc" ; import { register_listings_table } from "./listings" ; import { register_project_info_table } from "./project-info" ; import { register_project_status_table } from "./project-status" ; import { register_usage_info_table } from "./usage-info" ; type Query = { [ key ] : any ; } ; interface Spark { address : { ip : string ; } ; id : string ; conn : { id : string ; write : ( obj : any ) => boolean ; once : ( str : string , fn : Function ) => void ; on : ( str : string , fn : Function ) => void ; writable : boolean ; } ; write : ( obj : any ) => boolean ; end : ( ... args ) => void ; on : ( str : string , fn : Function ) => void ; } interface Channel { write : ( obj : any ) => boolean ; on : ( str : string , fn : Function ) => void ; forEach : ( fn : Function ) => void ; destroy : Function ; } import { Client } from "@cocalc/sync/editor/generic/types" ; interface Primus { channel : ( str : string ) => Channel ; } interface Logger { debug : Function ; } import stringify from "fast-json-stable-stringify" ; const { sha1 } = require ( "@cocalc/backend/misc_node" ) ; const COCALC_EPHEMERAL_STATE = process . env . COCALC_EPHEMERAL_STATE === "yes" ; class SyncTableChannel { private synctable : SyncTable ; private client : Client ; private logger : Logger ; public readonly name : string ; private query : Query ; private options : any [ ] = [ ] ; private query_string : string ; private channel : Channel ; private closed : boolean = false ; private num_connections : { n : number ; changed : Date ; } = { n : 0 , changed : new Date ( ) , } ; private ephemeral : boolean = false ; private persistent : boolean = false ; private connections_from_one_client : { [ id ] : number ; } = { } ; constructor ( { client , primus , query , options , logger , name , } ) { this . name = name ; this . client = client ; this . logger = logger ; this . query = query ; this . init_options ( options ) ; if ( COCALC_EPHEMERAL_STATE ) { this . ephemeral = true ; } this . query_string = stringify ( query ) ; this . channel = primus . channel ( this . name ) ; this . log ( `creating new sync channel (persistent=${ this . persistent }, ephemeral=${ this . ephemeral })` ) ; } public async init ( ) { this . init_handlers ( ) ; await this . init_synctable ( ) ; } private init_options ( options ) { if ( options == null ) { return ; } for ( const option of deep_copy ( options ) ) { if ( typeof option != "object" || option == null ) { throw Error ( "invalid options" ) ; } for ( const x of [ "ephemeral" , "persistent" ] ) { if ( option [ x ] != null ) { this [ x ] = option [ x ] ; delete option [ x ] ; } } if ( len ( option ) > 0 ) { this . options . push ( option ) ; } } } private log ( ... args ) { if ( this . logger == null ) return ; this . logger . debug ( `SyncTableChannel('${ this . name }', '${ this . query_string }'${ this . closed ? ",CLOSED" : "" }): ` , ... args ) ; } private init_handlers ( ) { this . log ( "init_handlers" ) ; this . channel . on ( "connection" , this . new_connection . bind ( this ) ) ; this . channel . on ( "disconnection" , this . end_connection . bind ( this ) ) ; } private async init_synctable ( ) { this . log ( "init_synctable" ) ; let create_synctable ; if ( this . ephemeral ) { this . log ( "init_synctable -- ephemeral (no database)" ) ; create_synctable = synctable_no_database ; } else { this . log ( "init_synctable -- persistent (but no changefeeds)" ) ; create_synctable = synctable_no_changefeed ; } this . synctable = create_synctable ( this . query , this . options , this . client ) ; this . synctable . once ( "closed" , this . close . bind ( this ) ) ; if ( this . query [ this . synctable . get_table ( ) ] [ 0 ] . string_id != null ) { register_synctable ( this . query , this . synctable ) ; } if ( this . synctable . table === "syncstrings" ) { this . log ( "init_synctable -- syncstrings: also initialize syncdoc..." ) ; init_syncdoc ( this . client , this . synctable , this . logger ) ; } this . synctable . on ( "versioned-changes" , this . send_versioned_changes_to_browsers . bind ( this ) ) ; this . log ( "created synctable -- waiting for connected state" ) ; await once ( this . synctable , "connected" ) ; this . log ( "created synctable -- now connected" ) ; this . broadcast_synctable_to_browsers ( ) ; } private increment_connection_count ( spark ) { let m = this . connections_from_one_client [ spark . conn . id ] ; if ( m === undefined ) m = 0 ; return ( this . connections_from_one_client [ spark . conn . id ] = m + 1 ) ; } private decrement_connection_count ( spark ) { const m = this . connections_from_one_client [ spark . conn . id ] ; if ( m === undefined ) { return 0 ; } return ( this . connections_from_one_client [ spark . conn . id ] = Math . max ( 0 , m - 1 ) ) ; } private async new_connection ( spark ) { const n = this . num_connections . n + 1 ; this . num_connections = { n , changed : new Date ( ) } ; const m = this . increment_connection_count ( spark ) ; this . log ( `new connection from (address=${ spark . address . ip }, conn=${ spark . conn . id }) -- ${ spark . id } -- num_connections = ${ n } (from this client = ${ m })` ) ; if ( m > MAX_CONNECTIONS_FROM_ONE_CLIENT ) { const error = `Too many connections (${ m } > ${ MAX_CONNECTIONS_FROM_ONE_CLIENT }) from this client.  You might need to refresh your browser.` ; this . log ( `${ error }  Waiting 15s, then killing new connection from ${ spark . id }...` ) ; await delay ( 15000 ) ; this . decrement_connection_count ( spark ) ; spark . end ( { error } ) ; return ; } if ( n > MAX_CONNECTIONS ) { const error = `Too many connections (${ n } > ${ MAX_CONNECTIONS })` ; this . log ( `${ error } Waiting 5s, then killing new connection from ${ spark . id }` ) ; await delay ( 5000 ) ; this . decrement_connection_count ( spark ) ; spark . end ( { error } ) ; return ; } if ( this . closed ) { this . log ( `table closed: killing new connection from ${ spark . id }` ) ; this . decrement_connection_count ( spark ) ; spark . end ( ) ; return ; } if ( this . synctable != null && this . synctable . get_state ( ) == "closed" ) { this . log ( `table state closed: killing new connection from ${ spark . id }` ) ; this . decrement_connection_count ( spark ) ; spark . end ( ) ; return ; } if ( this . synctable != null && this . synctable . get_state ( ) == "disconnected" ) { await once ( this . synctable , "connected" ) ; } this . send_synctable_to_browser ( spark ) ; spark . on ( "data" , async ( mesg ) => { try { await this . handle_mesg_from_browser ( spark , mesg ) ; } catch ( err ) { spark . write ( { error : `error handling mesg -- ${ err }` } ) ; this . log ( "error handling mesg -- " , err , err . stack ) ; } } ) ; } private async end_connection ( spark ) { const n = Math . max ( 0 , this . num_connections . n - 1 ) ; this . num_connections = { n , changed : new Date ( ) } ; const m = this . decrement_connection_count ( spark ) ; this . log ( `spark event -- end connection ${ spark . address . ip } -- ${ spark . id }  -- num_connections = ${ n }  (from this client = ${ m })` ) ; this . check_if_should_close ( ) ; } private send_synctable_to_browser ( spark ) { if ( this . closed ) return ; this . log ( "send_synctable_to_browser" ) ; spark . write ( { init : this . synctable . initial_version_for_browser_client ( ) } ) ; } private broadcast_synctable_to_browsers ( ) { if ( this . closed ) return ; this . log ( "broadcast_synctable_to_browsers" ) ; const x = { init : this . synctable . initial_version_for_browser_client ( ) } ; this . channel . write ( x ) ; } private check_if_should_close ( ) { if ( this . closed || this . persistent ) { return ; } const { n } = this . num_connections ; if ( n === 0 ) { this . log ( "check_if_should_close -- " , n , " -- do a save and maybe close" ) ; this . save_and_close_if_possible ( ) ; } else { this . log ( "check_if_should_close -- " , n , " -- do not close" ) ; } } private async handle_mesg_from_browser ( _spark , mesg ) { this . log ( "handle_mesg_from_browser " , ( this . channel as any ) . channel ) ; if ( this . closed ) { throw Error ( "received mesg from browser AFTER close" ) ; } if ( mesg == null ) { throw Error ( "mesg must not be null" ) ; } if ( mesg . timed_changes != null ) { this . synctable . apply_changes_from_browser_client ( mesg . timed_changes ) ; } await this . synctable . save ( ) ; } private send_versioned_changes_to_browsers ( versioned_changes ) { if ( this . closed ) return ; this . log ( "send_versioned_changes_to_browsers" ) ; const x = { versioned_changes } ; this . channel . write ( x ) ; } private async save_and_close_if_possible ( ) { if ( this . closed ) return ; this . log ( "save_and_close_if_possible: no connections, so saving..." ) ; await this . synctable . save ( ) ; const { n , changed } = this . num_connections ; const delay = new Date ( ) . valueOf ( ) - changed . valueOf ( ) ; this . log ( `save_and_close_if_possible: after save there are ${ n } connections and delay=${ delay }` ) ; if ( n === 0 ) { if ( delay < CLOSE_DELAY_MS ) { this . log ( `save_and_close_if_possible: wait a bit then try again` ) ; setTimeout ( this . check_if_should_close . bind ( this ) , 1000 + CLOSE_DELAY_MS - delay ) ; } else { this . log ( `save_and_close_if_possible: close this SyncTableChannel atomically` ) ; this . close ( ) ; } } else { this . log ( `save_and_close_if_possible: NOT closing this SyncTableChannel` ) ; } } private close ( ) { if ( this . closed ) { return ; } this . log ( "close: closing" ) ; delete synctable_channels [ this . name ] ; this . channel . destroy ( ) ; this . synctable . close_no_async ( ) ; this . log ( "close: closed" ) ; close ( this ) ; this . closed = true ; } public get_synctable ( ) { return this . synctable ; } } const synctable_channels = { } ; function createKey ( args ) { return stringify ( [ args [ 3 ] , args [ 4 ] ] ) ; } function channel_name ( query , options ) { const opts = { } ; for ( const x of options ) { for ( const key in x ) { opts [ key ] = x [ key ] ; } } let q ; try { q = key ( query ) ; } catch { q = query ; } const y = stringify ( [ q , opts ] ) ; const s = sha1 ( y ) ; return `sync:${ s }` ; } async function synctable_channel0 ( client , primus , logger , query , options ) { const name = channel_name ( query , options ) ; logger . debug ( "synctable_channel" , JSON . stringify ( query ) , name ) ; if ( query ?. syncstrings != null ) { const path = query ?. syncstrings [ 0 ] ?. path ; if ( client . is_deleted ( path ) ) { logger . debug ( `synctable_channel -- refusing to open "${ path }" since it is marked as deleted` ) ; throw Error ( `${ path } is deleted` ) ; } } if ( synctable_channels [ name ] === undefined ) { synctable_channels [ name ] = new SyncTableChannel ( { client , primus , name , query , options , logger , } ) ; await synctable_channels [ name ] . init ( ) ; if ( query ?. listings != null ) { register_listings_table ( synctable_channels [ name ] . get_synctable ( ) , logger , client . client_id ( ) ) ; } else if ( query ?. project_info != null ) { register_project_info_table ( synctable_channels [ name ] . get_synctable ( ) , logger , client . client_id ( ) ) ; } else if ( query ?. project_status != null ) { register_project_status_table ( synctable_channels [ name ] . get_synctable ( ) , logger , client . client_id ( ) ) ; } else if ( query ?. usage_info != null ) { register_usage_info_table ( synctable_channels [ name ] . get_synctable ( ) , client . client_id ( ) ) ; } } return name ; } export const synctable_channel = reuseInFlight ( synctable_channel0 , { createKey , } ) ;