import { EventEmitter } from "events" ; import * as misc from "@cocalc/util/misc" ; import { callback } from "awaiting" ; import { PostgreSQL , QuerySelect } from "./types" ; import { query } from "./changefeed-query" ; type WhereCondition = Function | object | object [ ] ; type ChangeAction = "delete" | "insert" | "update" ; function parse_action ( obj ) { const s = `${ obj . toLowerCase ( ) }` ; if ( s === "delete" || s === "insert" || s === "update" ) { return s ; } throw Error ( `invalid action "${ s }"` ) ; } export interface ChangeEvent { action : ChangeAction ; new_val ? : object ; old_val ? : object ; } export class Changes extends EventEmitter { private db : PostgreSQL ; private table : string ; private select : QuerySelect ; private watch : string [ ] ; private where : WhereCondition ; private trigger_name : string ; private closed : boolean ; private condition ? : { [ field ] : Function ; } ; private match_condition : Function ; private val_update_cache : { [ key ] : any ; } = { } ; constructor ( db , table , select , watch , where , cb ) { super ( ) ; this . handle_change = this . handle_change . bind ( this ) ; this . db = db ; this . table = table ; this . select = select ; this . watch = watch ; this . where = where ; this . init ( cb ) ; } async init ( cb ) { this . dbg ( "constructor" ) ( `select=${ misc . to_json ( this . select ) }, watch=${ misc . to_json ( this . watch ) }, @_where=${ misc . to_json ( this . where ) }` ) ; try { this . init_where ( ) ; } catch ( e ) { cb ( `error initializing where conditions -- ${ e }` ) ; return ; } try { this . trigger_name = await callback ( this . db . _listen , this . table , this . select , this . watch ) ; } catch ( err ) { cb ( err ) ; return ; } this . db . on ( this . trigger_name , this . handle_change ) ; this . db . setMaxListeners ( 0 ) ; this . db . once ( "connect" , this . close ) ; cb ( undefined , this ) ; } private dbg ( f ) { return this . db . _dbg ( `Changes(table='${ this . table }').${ f }` ) ; } private fail ( err ) { if ( this . closed ) { return ; } this . dbg ( "_fail" ) ( `err='${ err }'` ) ; this . emit ( "error" , new Error ( err ) ) ; this . close ( ) ; } public close ( ) { if ( this . closed ) { return ; } this . emit ( "close" , { action : "close" } ) ; this . removeAllListeners ( ) ; if ( this . db != null ) { this . db . removeListener ( this . trigger_name , this . handle_change ) ; this . db . removeListener ( "connect" , this . close ) ; this . db . _stop_listening ( this . table , this . select , this . watch ) ; } misc . close ( this ) ; this . closed = true ; } public async insert ( where ) { const where0 = { } ; for ( const k in where ) { const v = where [ k ] ; where0 [ `${ k } = $` ] = v ; } let results ; try { results = await query ( { db : this . db , select : this . watch . concat ( misc . keys ( this . select ) ) , table : this . table , where : where0 , one : false , } ) ; } catch ( err ) { this . fail ( err ) ; return ; } for ( const x of results ) { if ( this . match_condition ( x ) ) { misc . map_mutate_out_undefined ( x ) ; const change = { action : "insert" , new_val : x } ; this . emit ( "change" , change ) ; } } } public delete ( where ) { const change = { action : "delete" , old_val : where } ; this . emit ( "change" , change ) ; } private async handle_change ( mesg ) { if ( this . closed ) { return ; } if ( mesg [ 0 ] === "DELETE" ) { if ( ! this . match_condition ( mesg [ 2 ] ) ) { return ; } this . emit ( "change" , { action : "delete" , old_val : mesg [ 2 ] } ) ; return ; } let k , r , v ; if ( typeof mesg [ 0 ] !== "string" ) { throw Error ( `invalid mesg -- mesg[0] must be a string` ) ; } let action = parse_action ( mesg [ 0 ] ) ; if ( ! this . match_condition ( mesg [ 1 ] ) ) { if ( action !== "update" ) { return ; } for ( k in mesg [ 1 ] ) { v = mesg [ 1 ] [ k ] ; if ( mesg [ 2 ] [ k ] == null ) { mesg [ 2 ] [ k ] = v ; } } if ( this . match_condition ( mesg [ 2 ] ) ) { this . emit ( "change" , { action : "delete" , old_val : mesg [ 2 ] } ) ; } return ; } if ( this . watch . length === 0 ) { r = { action , new_val : mesg [ 1 ] } ; this . emit ( "change" , r ) ; return ; } const where = { } ; for ( k in mesg [ 1 ] ) { v = mesg [ 1 ] [ k ] ; where [ `${ k } = $` ] = v ; } let result ; try { result = await query ( { db : this . db , select : this . watch , table : this . table , where , one : true , } ) ; } catch ( err ) { this . fail ( err ) ; return ; } if ( this . closed ) { return ; } if ( result == null ) { this . emit ( "change" , { action : "delete" , old_val : mesg [ 1 ] } ) ; return ; } const key = JSON . stringify ( mesg [ 1 ] ) ; const this_val = misc . merge ( result , mesg [ 1 ] ) ; let new_val ; if ( action == "update" ) { const x = this . new_val_update ( mesg [ 1 ] , this_val , key ) ; if ( x == null ) { return ; } action = x . action ; new_val = x . new_val ; } else { action = "insert" ; new_val = this_val ; } this . val_update_cache [ key ] = this_val ; r = { action , new_val } ; this . emit ( "change" , r ) ; } private new_val_update ( primary_part , this_val , key ) { if ( this . closed ) { return ; } const prev_val = this . val_update_cache [ key ] ; if ( prev_val == null ) { return { new_val : this_val , action : "insert" } ; } const new_val = misc . copy ( primary_part ) ; for ( const field in this_val ) { if ( new_val [ field ] === undefined && JSON . stringify ( this_val [ field ] ) != JSON . stringify ( prev_val [ field ] ) ) { new_val [ field ] = this_val [ field ] ; } } return { new_val , action : "update" } ; } private init_where ( ) { if ( typeof this . where === "function" ) { this . match_condition = this . where ; return ; } let w ; if ( misc . is_object ( this . where ) ) { w = [ this . where ] ; } else { w = this . where as object [ ] ; } this . condition = { } ; const add_condition = ( field , op , val ) : void => { if ( this . condition == null ) return ; let f , g ; field = field . trim ( ) ; if ( field [ 0 ] === '"' ) { field = field . slice ( 1 , field . length - 1 ) ; } if ( this . select [ field ] == null ) { throw Error ( `'${ field }' must be in select="${ JSON . stringify ( this . select ) }"` ) ; } if ( misc . is_object ( val ) ) { throw Error ( `val (=${ misc . to_json ( val ) }) must not be an object` ) ; } if ( misc . is_array ( val ) ) { if ( op === "=" || op === "==" ) { f = function ( x ) { for ( const v of val ) { if ( x === v ) { return true ; } } return false ; } ; } else if ( op === "!=" || op === "<>" ) { f = function ( x ) { for ( const v of val ) { if ( x === v ) { return false ; } } return true ; } ; } else { throw Error ( "if val is an array, then op must be = or !=" ) ; } } else if ( misc . is_date ( val ) ) { if ( op == "=" || op == "==" ) { f = ( x ) => new Date ( x ) . valueOf ( ) - val === 0 ; } else if ( op == "!=" || op == "<>" ) { f = ( x ) => new Date ( x ) . valueOf ( ) - val !== 0 ; } else { g = misc . op_to_function ( op ) ; f = ( x ) => g ( new Date ( x ) , val ) ; } } else { g = misc . op_to_function ( op ) ; f = ( x ) => g ( x , val ) ; } this . condition [ field ] = f ; } ; for ( const obj of w ) { let found , i , op ; if ( misc . is_object ( obj ) ) { for ( const k in obj ) { const val = obj [ k ] ; found = false ; for ( op of misc . operators ) { i = k . indexOf ( op ) ; if ( i !== - 1 ) { add_condition ( k . slice ( 0 , i ) . trim ( ) , op , val ) ; found = true ; break ; } } if ( ! found ) { throw Error ( `unable to parse '${ k }'` ) ; } } } else if ( typeof obj === "string" ) { found = false ; for ( op of misc . operators ) { i = obj . indexOf ( op ) ; if ( i !== - 1 ) { add_condition ( obj . slice ( 0 , i ) , op , eval ( obj . slice ( i + op . length ) . trim ( ) ) ) ; found = true ; break ; } } if ( ! found ) { throw Error ( `unable to parse '${ obj }'` ) ; } } else { throw Error ( "NotImplementedError" ) ; } } if ( misc . len ( this . condition ) === 0 ) { delete this . condition ; } this . match_condition = ( obj ) : boolean => { if ( this . condition == null ) { return true ; } for ( const field in this . condition ) { const f = this . condition [ field ] ; if ( ! f ( obj [ field ] ) ) { return false ; } } return true ; } ; } }