import { AsyncEventSubscription , asyncOperation , ClassType } from '@deepkit/core' ; import { AsyncSubscription } from '@deepkit/core-rxjs' ; import { BaseQuery , Database , DatabaseQueryModel , DatabaseRegistry , exportQueryFilterFieldNames , FilterQuery , QueryDatabaseDeleteEvent , replaceQueryFilterParameter , Sort , UnitOfWorkEvent , UnitOfWorkUpdateEvent } from '@deepkit/orm' ; import { Collection , CollectionSort , EntitySubject , IdVersionInterface , rpcEntityPatch , rpcEntityRemove , RpcKernelBaseConnection , RpcTypes } from '@deepkit/rpc' ; import { ClassSchema , getClassSchema , resolveClassTypeOrForward } from '@deepkit/type' ; import { Observable } from 'rxjs' ; import { Broker , EntityChannelMessageType , EntityPatches } from '../broker/broker' ; import { injectable } from '@deepkit/injector' ; import { findQuerySatisfied } from '../utils' ; interface SentState { lastSentVersion ? : number ; listeners : number ; } class SubscriptionHandler { protected sentEntities : { [ id ] : SentState ; } = { } ; protected entitySubscription ? : Promise < AsyncSubscription > ; constructor ( protected connection , protected classSchema , protected database , protected broker ) { } protected hasSentState ( id ) { return ! ! this . sentEntities [ id ] ; } /**
     * Necessary when the whole state of `id` should be deleted from memory, so it wont sync to client anymore.
     */ protected rmSentState < T > ( id ) { delete this . sentEntities [ id ] ; if ( Object . keys ( this . sentEntities ) . length === 0 ) { if ( this . entitySubscription ) { this . entitySubscription . then ( v => v . unsubscribe ( ) ) ; this . entitySubscription = undefined ; } } } protected getSentState ( id ) { if ( ! this . sentEntities [ id ] ) { this . sentEntities [ id ] = { lastSentVersion : 0 , listeners : 0 , } ; } return this . sentEntities [ id ] ; } public setSent ( id , version ? ) { this . getSentState ( id ) . lastSentVersion = version ; } public needsToBeSend ( id , version ) { if ( ! this . hasSentState ( id ) ) return false ; const state = this . getSentState ( id ) ; return state . listeners > 0 && ( state . lastSentVersion === undefined || ( version === 0 || version > state . lastSentVersion ) ) ; } public decreaseUsage ( id ) { const state = this . getSentState ( id ) ; state . listeners -- ; if ( state . listeners <= 0 ) { this . rmSentState ( id ) ; } } public increaseUsage ( id ) { const state = this . getSentState ( id ) ; state . listeners ++ ; this . subscribeEntity ( ) ; } subscribeEntity ( ) { if ( this . entitySubscription ) { return ; } const entityName = this . classSchema . getName ( ) ; this . entitySubscription = this . broker . entityChannel ( this . classSchema ) . subscribe ( ( message ) => { if ( message . type === EntityChannelMessageType . remove ) { for ( const id of message . ids ) { this . rmSentState ( id ) ; } this . connection . createMessageBuilder ( ) . composite ( RpcTypes . Entity ) . add ( RpcTypes . EntityRemove , rpcEntityRemove , { entityName : entityName , ids : message . ids , } ) . send ( ) ; return ; } if ( message . type === EntityChannelMessageType . patch && this . needsToBeSend ( message . id , message . version ) ) { this . setSent ( message . id , message . version ) ; this . connection . createMessageBuilder ( ) . composite ( RpcTypes . Entity ) . add ( RpcTypes . EntityPatch , rpcEntityPatch , { entityName : entityName , id : message . id , version : message . version , patch : message . patch } ) . send ( ) ; } } ) ; } } class SubscriptionHandlers { protected handler = new Map < ClassSchema , SubscriptionHandler > ( ) ; constructor ( protected connection , protected databases , protected broker ) { } get ( classSchema ) { let handler = this . handler . get ( classSchema ) ; if ( ! handler ) { handler = new SubscriptionHandler ( this . connection , classSchema , this . databases . getDatabaseForEntity ( classSchema ) , this . broker ) ; this . handler . set ( classSchema , handler ) ; } return handler ; } } class LiveDatabaseQueryModel < T > extends DatabaseQueryModel < T , FilterQuery < T > , Sort < T , any > > { public disableEntityChangeFeed = false ; isChangeFeedActive ( ) { return ! this . disableEntityChangeFeed ; } getCollectionSort ( ) { const sort = [ ] ; if ( ! this . sort ) return sort ; for ( const [ name , direction ] of Object . entries ( this . sort as Sort < T > ) ) { sort . push ( { field : name , direction : direction as 'asc' | 'desc' } ) ; } return sort ; } } type JoinedClassSchemaInfo = { classSchema : ClassSchema ; fields : string [ ] ; filter : FilterQuery < any > ; usedEntityFieldsSubscription ? : AsyncSubscription ; entitySubscription ? : AsyncSubscription ; } ; function extractModelData ( classSchema , model , result ) { result . push ( { classSchema , filter : model . filter || { } , fields : model . filter ? exportQueryFilterFieldNames ( classSchema , model . filter ) : [ ] , } ) ; extractJoinedClassSchemaInfos ( model , result ) ; } function extractJoinedClassSchemaInfos ( model , result ) { for ( const join of model . joins ) { if ( join . propertySchema . backReference && join . propertySchema . backReference . via ) { const schema = getClassSchema ( resolveClassTypeOrForward ( join . propertySchema . backReference . via ) ) ; result . push ( { classSchema : schema , filter : { } , fields : [ ] , } ) ; } extractModelData ( join . query . classSchema , join . query . model , result ) ; } } class LiveCollection < T extends IdVersionInterface > { protected joinedClassSchemas : JoinedClassSchemaInfo [ ] = [ ] ; protected rootFields : string [ ] = [ ] ; protected rootUsedEntityFieldsSubscription ? : AsyncSubscription ; protected entitySubscription ? : AsyncSubscription ; protected knownIDs = new Set < string | number > ( ) ; protected classSchema = getClassSchema ( this . collection . classType ) ; constructor ( protected collection , protected model , protected broker , protected database , protected subscriptionHandler , protected connection ) { this . rootFields = exportQueryFilterFieldNames ( this . classSchema , model . filter || { } ) ; extractJoinedClassSchemaInfos ( this . model , this . joinedClassSchemas ) ; } protected async publishUseEntityFields ( ) { const promises = [ ] ; if ( this . rootFields . length ) promises . push ( this . broker . publishEntityFields ( this . classSchema , this . rootFields ) . then ( v => this . rootUsedEntityFieldsSubscription = v ) ) ; for ( const info of this . joinedClassSchemas ) { if ( ! info . fields . length ) continue ; promises . push ( this . broker . publishEntityFields ( info . classSchema , info . fields ) . then ( sub => { info . usedEntityFieldsSubscription = sub ; } ) ) ; } await Promise . all ( promises ) ; } public async stopSync ( ) { if ( this . entitySubscription ) this . entitySubscription . unsubscribe ( ) ; const promises = [ ] ; for ( const [ , info ] of this . joinedClassSchemas . entries ( ) ) { if ( info . usedEntityFieldsSubscription ) promises . push ( info . usedEntityFieldsSubscription . unsubscribe ( ) ) ; if ( info . entitySubscription ) info . entitySubscription . unsubscribe ( ) ; } if ( this . rootUsedEntityFieldsSubscription ) promises . push ( this . rootUsedEntityFieldsSubscription . unsubscribe ( ) ) ; await Promise . all ( promises ) ; this . joinedClassSchemas = [ ] ; for ( const id of this . knownIDs . values ( ) ) { if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . decreaseUsage ( id ) ; } } protected lastUpdatePromise : any ; protected async updateCollection ( databaseChanged = false ) { if ( this . lastUpdatePromise ) return this . lastUpdatePromise ; this . lastUpdatePromise = asyncOperation ( ( resolve ) => { setTimeout ( async ( ) => { await this . _updateCollection ( databaseChanged ) ; this . lastUpdatePromise = undefined ; resolve ( undefined ) ; } , 100 ) ; } ) ; return this . lastUpdatePromise ; } protected async _updateCollection ( databaseChanged = false ) { if ( ! this . entitySubscription ) return ; const query = this . database . query ( this . classSchema ) ; query . model = this . model . clone ( query ) ; query . model . limit = undefined ; query . model . skip = undefined ; const total = await query . count ( ) ; query . model = this . model ; const items = await query . find ( ) ; const copiedKnownIds = new Set ( this . knownIDs . values ( ) ) ; for ( const item of items ) { copiedKnownIds . delete ( item . id ) ; if ( ! this . knownIDs . has ( item . id ) ) { this . knownIDs . add ( item . id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . increaseUsage ( item . id ) ; this . collection . add ( item ) ; } } for ( const id of copiedKnownIds . values ( ) ) { this . knownIDs . delete ( id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . decreaseUsage ( id ) ; } const idsToRemove = [ ... copiedKnownIds . values ( ) ] ; if ( idsToRemove . length > 0 ) this . collection . removeMany ( idsToRemove ) ; this . collection . setSort ( items . map ( v => v . id ) ) ; if ( this . collection . state . total !== total ) { this . collection . state . total = total ; this . collection . setState ( this . collection . state ) ; } } protected async getItem ( id ) { const query = this . database . query ( this . classSchema ) ; query . model = this . model . clone ( query ) ; query . filter ( { id : id } ) ; return query . findOneOrUndefined ( ) ; } public async startSync ( ) { if ( this . entitySubscription ) throw new Error ( 'Collection sync already started' ) ; await this . publishUseEntityFields ( ) ; for ( const info of this . joinedClassSchemas ) { if ( info . entitySubscription ) continue ; info . entitySubscription = await this . broker . entityChannel ( info . classSchema ) . subscribe ( ( message ) => { this . updateCollection ( true ) ; } ) ; } let currentQuery = replaceQueryFilterParameter ( this . classSchema , this . model . filter || { } , this . model . parameters ) ; function requiresFullUpdate ( collection ) { return collection . model . hasPaging ( ) || collection . model . hasSort ( ) ; } this . entitySubscription = await this . broker . entityChannel ( this . classSchema ) . subscribe ( async ( message ) => { if ( message . type === EntityChannelMessageType . remove ) { if ( requiresFullUpdate ( this . collection ) ) { await this . updateCollection ( true ) ; return ; } for ( const id of message . ids ) { this . knownIDs . delete ( id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . decreaseUsage ( id ) ; } this . collection . removeMany ( message . ids ) ; } if ( message . type === EntityChannelMessageType . add ) { if ( this . knownIDs . has ( message . id ) || ! findQuerySatisfied ( message . item , currentQuery ) ) { return ; } if ( requiresFullUpdate ( this . collection ) ) { await this . updateCollection ( true ) ; return ; } this . knownIDs . add ( message . id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . increaseUsage ( message . id ) ; this . collection . add ( message . item ) ; } if ( message . type === EntityChannelMessageType . patch ) { const querySatisfied = findQuerySatisfied ( message . item , currentQuery ) ; if ( this . knownIDs . has ( message . id ) && ! querySatisfied ) { if ( requiresFullUpdate ( this . collection ) ) { await this . updateCollection ( true ) ; } else { this . knownIDs . delete ( message . id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . decreaseUsage ( message . id ) ; this . collection . remove ( message . id ) ; } } else if ( ! this . knownIDs . has ( message . id ) && querySatisfied ) { if ( requiresFullUpdate ( this . collection ) ) { await this . updateCollection ( true ) ; } else { this . knownIDs . add ( message . id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . increaseUsage ( message . id ) ; const item = await this . getItem ( message . id ) ; if ( item ) { this . collection . add ( item ) ; } else { await this . updateCollection ( true ) ; } } } } } ) ; this . model . change . subscribe ( ( ) => { this . updateCollection ( ) . catch ( console . error ) ; } ) ; this . collection . model . change . subscribe ( ( ) => { this . model . limit = this . collection . model . limit ; this . model . skip = this . collection . model . skip ; this . model . itemsPerPage = this . collection . model . itemsPerPage ; this . model . sort = this . collection . model . sort as Sort < any > ; if ( this . collection . model . filter ) this . model . filter = this . collection . model . filter ; this . model . setParameters ( this . collection . model . parameters ) ; this . updateCollection ( ) . catch ( console . error ) ; } ) ; const query = this . database . query ( this . classSchema ) ; query . model = this . model . clone ( query ) ; query . model . limit = undefined ; query . model . skip = undefined ; const total = await query . count ( ) ; this . collection . state . total = total ; query . model = this . model ; const items = await query . find ( ) ; for ( const item of items ) { this . knownIDs . add ( item . id ) ; if ( this . model . isChangeFeedActive ( ) ) this . subscriptionHandler . increaseUsage ( item . id ) ; } this . collection . set ( items ) ; } } export class LiveQuery < T extends IdVersionInterface > extends BaseQuery < T > { public model = this . createModel < T > ( ) ; constructor ( public classSchema , protected connection , protected database , protected broker , protected subscriptionHandler ) { super ( classSchema ) ; } clone ( ) { const cloned = new ( this [ 'constructor' ] as ClassType < this > ) ( this . classSchema , this . connection , this . database , this . broker , this . subscriptionHandler ) ; cloned . model = this . model . clone ( cloned ) as this [ 'model' ] ; return cloned ; } disableEntityChangeFeed ( ) { this . model . disableEntityChangeFeed = true ; return this ; } protected createModel < T > ( ) { return new LiveDatabaseQueryModel ( ) ; } async findOneOrUndefined ( ) { const item = await this . database . query ( this . classSchema ) . filter ( this . model . filter ) . findOneOrUndefined ( ) ; if ( ! item ) return ; const foundId = item . id ; this . subscriptionHandler . increaseUsage ( foundId ) ; this . subscriptionHandler . setSent ( item . id , 0 ) ; this . subscriptionHandler . subscribeEntity ( ) ; return new EntitySubject ( item , ( ) => { this . subscriptionHandler . decreaseUsage ( foundId ) ; } ) ; } async findOne ( ) { const item = await this . database . query ( this . classSchema ) . filter ( this . model . filter ) . findOne ( ) ; const foundId = item . id ; this . subscriptionHandler . increaseUsage ( foundId ) ; this . subscriptionHandler . setSent ( item . id , 0 ) ; this . subscriptionHandler . subscribeEntity ( ) ; return new EntitySubject ( item , ( ) => { this . subscriptionHandler . decreaseUsage ( foundId ) ; } ) ; } public count < T extends IdVersionInterface > ( ) { const rootFields = exportQueryFilterFieldNames ( this . classSchema , this . model . filter || { } ) ; let currentQuery = replaceQueryFilterParameter ( this . classSchema , this . model . filter || { } , this . model . parameters ) ; const joinedClassSchemas = [ ] ; extractJoinedClassSchemaInfos ( this . model , joinedClassSchemas ) ; return new Observable ( ( observer ) => { let rootEntitySub ; let usedEntityFieldsSub ; ( async ( ) => { usedEntityFieldsSub = await this . broker . publishEntityFields ( this . classSchema , rootFields ) ; const knownIDs = new Set < string | number > ( ) ; let counter = 0 ; rootEntitySub = await this . broker . entityChannel ( this . classSchema ) . subscribe ( ( message ) => { if ( message . type === EntityChannelMessageType . add ) { if ( ! knownIDs . has ( message . id ) && findQuerySatisfied ( message . item , currentQuery ) ) { counter ++ ; knownIDs . add ( message . id ) ; observer . next ( counter ) ; } } if ( message . type === EntityChannelMessageType . patch ) { if ( knownIDs . has ( message . id ) && ! findQuerySatisfied ( message . item , currentQuery ) ) { counter -- ; knownIDs . delete ( message . id ) ; observer . next ( counter ) ; } else if ( ! knownIDs . has ( message . id ) && findQuerySatisfied ( message . item , currentQuery ) ) { counter ++ ; knownIDs . add ( message . id ) ; observer . next ( counter ) ; } } if ( message . type === EntityChannelMessageType . remove ) { for ( const id of message . ids ) { if ( knownIDs . has ( id ) ) { counter -- ; knownIDs . delete ( id ) ; observer . next ( counter ) ; } } } } ) ; const updateAll = async ( ) => { const query = this . database . query ( this . classSchema ) ; query . model = this . model . clone ( query ) ; const items = await query . select ( 'id' ) . find ( ) ; knownIDs . clear ( ) ; counter = items . length ; for ( const item of items ) knownIDs . add ( item . id ) ; observer . next ( counter ) ; } ; for ( const info of joinedClassSchemas ) { if ( info . entitySubscription ) continue ; info . entitySubscription = await this . broker . entityChannel ( info . classSchema ) . subscribe ( ( message ) => { updateAll ( ) ; } ) ; } await updateAll ( ) ; } ) ( ) ; return { unsubscribe : async ( ) => { rootEntitySub . unsubscribe ( ) ; await usedEntityFieldsSub . unsubscribe ( ) ; for ( const info of joinedClassSchemas ) { if ( info . entitySubscription ) info . entitySubscription . unsubscribe ( ) ; } } } ; } ) ; } async find ( ) { const collection = new Collection ( this . classSchema . classType ) ; collection . model . set ( this . model ) ; const liveCollection = new LiveCollection ( collection , this . model , this . broker , this . database , this . subscriptionHandler , this . connection ) ; collection . addTeardown ( async ( ) => { await liveCollection . stopSync ( ) ; } ) ; await liveCollection . startSync ( ) ; return collection ; } /**
     * Returns a new Observable that resolves the id as soon as an item in the database of given filter criteria is found.
     */ public onCreation < T extends IdVersionInterface > ( initialCheck = true , stopOnFind = true ) { return new Observable ( ( observer ) => { let sub ; const filter = this . model . filter || { } ; ( async ( ) => { sub = await this . broker . entityChannel ( this . classSchema ) . subscribe ( ( message ) => { if ( message . type === EntityChannelMessageType . add && findQuerySatisfied ( message . item , filter ) ) { observer . next ( message . id ) ; if ( stopOnFind && sub ) sub . unsubscribe ( ) ; } } ) ; if ( initialCheck ) { const item = await this . database . query ( this . classSchema ) . select ( 'id' ) . filter ( filter ) . findOneOrUndefined ( ) ; if ( item ) { observer . next ( item . id ) ; if ( stopOnFind ) sub . unsubscribe ( ) ; } } } ) ( ) ; return { unsubscribe ( ) { if ( sub ) sub . unsubscribe ( ) ; } } ; } ) ; } } @ injectable export class LiveDatabase { protected subscriptionHandler = new SubscriptionHandlers ( this . connection , this . databases , this . broker ) ; protected entitySubscriptions = new Map < ClassSchema , AsyncEventSubscription [ ] > ( ) ; constructor ( protected databases , protected broker , protected connection ) { } public disableChangeFeed < T extends IdVersionInterface > ( classType ) { const schema = getClassSchema ( classType ) ; const subscriptions = this . entitySubscriptions . get ( schema ) ; if ( ! subscriptions ) return ; for ( const sub of subscriptions ) sub . unsubscribe ( ) ; this . entitySubscriptions . delete ( schema ) ; } public enableChangeFeed ( ... classTypes ) { for ( const classType of classTypes ) { this . setupListeners ( classType ) ; } } public getSubscriptionHandler ( classType ) { return this . subscriptionHandler . get ( getClassSchema ( classType ) ) ; } protected setupListeners ( classType ) { const database = this . databases . getDatabaseForEntity ( classType ) ; const schema = getClassSchema ( classType ) ; if ( this . entitySubscriptions . has ( schema ) ) return ; const subscriptions = [ ] ; subscriptions . push ( database . unitOfWorkEvents . onInsertPost . subscribe ( ( event ) => { if ( schema !== event . classSchema ) return ; for ( const item of event . items ) { this . broker . entityChannel ( event . classSchema ) . publishAdd ( item ) . catch ( ( ) => { console . warn ( 'Failed publishing add event' , schema . getClassName ( ) , item ) ; } ) ; } } ) ) ; subscriptions . push ( database . unitOfWorkEvents . onUpdatePre . subscribe ( ( event ) => { if ( schema !== event . classSchema ) return ; for ( const changeSet of event . changeSets ) { changeSet . changes . increase ( 'version' , 1 ) ; } } ) ) ; subscriptions . push ( database . unitOfWorkEvents . onUpdatePost . subscribe ( async ( event ) => { if ( schema !== event . classSchema ) return ; for ( const changeSet of event . changeSets ) { const jsonPatch = { $set : changeSet . changes . $set , $inc : changeSet . changes . $inc , $unset : changeSet . changes . $unset , } ; const fields = { } ; for ( const field of await this . broker . getEntityFields ( event . classSchema ) ) { fields [ field ] = ( changeSet . item as any ) [ field ] ; } this . broker . entityChannel ( event . classSchema ) . publishPatch ( changeSet . item . id , changeSet . item . version , jsonPatch , fields ) ; } } ) ) ; subscriptions . push ( database . unitOfWorkEvents . onDeletePost . subscribe ( ( event ) => { if ( schema !== event . classSchema ) return ; const ids = [ ] ; for ( const item of event . items ) ids . push ( item . id ) ; this . broker . entityChannel ( event . classSchema ) . publishRemove ( ids ) . catch ( ( ) => { console . warn ( 'Failed publishing remove event' , schema . getClassName ( ) , ids ) ; } ) ; ; } ) ) ; subscriptions . push ( database . queryEvents . onDeletePost . subscribe ( ( event ) => { if ( schema !== event . classSchema ) return ; this . broker . entityChannel ( event . classSchema ) . publishRemove ( event . deleteResult . primaryKeys ) . catch ( ( ) => { console . warn ( 'Failed publishing remove event' , schema . getClassName ( ) , event . deleteResult . primaryKeys ) ; } ) ; } ) ) ; subscriptions . push ( database . queryEvents . onPatchPre . subscribe ( async ( event ) => { if ( schema !== event . classSchema ) return ; event . patch . increase ( 'version' , 1 ) ; event . returning . push ( 'version' ) ; for ( const field of await this . broker . getEntityFields ( event . classSchema ) ) { if ( ! event . returning . includes ( field ) ) event . returning . push ( field ) ; } } ) ) ; subscriptions . push ( database . queryEvents . onPatchPost . subscribe ( async ( event ) => { if ( schema !== event . classSchema ) return ; const jsonPatch = { $set : event . patch . $set , $inc : event . patch . $inc , $unset : event . patch . $unset , } ; for ( let i = 0 ; i < event . patchResult . primaryKeys . length ; i ++ ) { const fields = { } ; for ( const field of await this . broker . getEntityFields ( event . classSchema ) ) { if ( ! ( event . patchResult . returning as any ) [ field ] ) continue ; fields [ field ] = ( event . patchResult . returning as any ) [ field ] [ i ] ; } this . broker . entityChannel ( event . classSchema ) . publishPatch ( event . patchResult . primaryKeys [ i ] , ( event . patchResult . returning as any ) [ 'version' ] [ i ] , jsonPatch , fields ) ; } } ) ) ; this . entitySubscriptions . set ( schema , subscriptions ) ; } public query < T extends IdVersionInterface > ( classType ) { return new LiveQuery ( getClassSchema ( classType ) , this . connection , this . databases . getDatabaseForEntity ( classType ) , this . broker , this . subscriptionHandler . get ( getClassSchema ( classType ) ) ) ; } }