import { Inverse , invertQuery } from '../query/inverter' ; import { Query } from '../query/query' ; import { FactEnvelope , FactPath , FactRecord , FactReference , Storage } from '../storage' ; import { mapAsync } from '../util/fn' ; import { Feed , Handler , Observable , Subscription } from './feed' ; type Listener = { inverse : Inverse ; match : FactReference ; added : Handler ; removed : Handler ; } ; class SubscriptionImpl implements Subscription { private listeners : Listener [ ] ; private loading : Promise < void > ; constructor ( private observable , private added , private removed , private results ) { this . listeners = observable . inverses . map ( inverse => { return { inverse : inverse , match : observable . start , added : added , removed : removed } ; } ) ; this . loading = this . beginLoading ( ) ; } add ( ) { this . listeners . forEach ( listener => { this . observable . addListener ( listener ) ; } ) ; } load ( ) { return this . loading ; } dispose ( ) { this . listeners . forEach ( listener => { this . observable . removeListener ( listener ) ; } ) ; } private async beginLoading ( ) { const paths = await this . results ; if ( paths . length > 0 ) { await this . added ( paths ) ; } } } class ObservableImpl implements Observable { constructor ( public start , public query , public inverses , public results , public addListener , public removeListener ) { } subscribe ( added , removed ) { const subscription = new SubscriptionImpl ( this , added , removed , this . results ) ; subscription . add ( ) ; return subscription ; } } export class FeedImpl implements Feed { private listenersByTypeAndQuery : { [ appliedToType ] : { [ queryKey ] : Listener [ ] ; } ; } ; constructor ( private inner ) { this . listenersByTypeAndQuery = { } ; } async save ( envelopes ) { const saved = await this . inner . save ( envelopes ) ; await mapAsync ( saved , async ( envelope ) => { await this . notifyFactSaved ( envelope . fact ) ; } ) ; return saved ; } query ( start , query ) { return this . inner . query ( start , query ) ; } exists ( fact ) { return this . inner . exists ( fact ) ; } load ( references ) { return this . inner . load ( references ) ; } from ( fact , query ) { const inverses = invertQuery ( query ) ; const observable = new ObservableImpl ( fact , query , inverses , this . inner . query ( fact , query ) , listener => { this . addListener ( listener ) ; } , listener => { this . removeListener ( listener ) ; } ) ; return observable ; } private addListener ( listener ) { let listenersByQuery = this . listenersByTypeAndQuery [ listener . inverse . appliedToType ] ; if ( ! listenersByQuery ) { listenersByQuery = { } ; this . listenersByTypeAndQuery [ listener . inverse . appliedToType ] = listenersByQuery ; } const queryKey = listener . inverse . affected . toDescriptiveString ( ) ; let listeners = listenersByQuery [ queryKey ] ; if ( ! listeners ) { listeners = [ ] ; listenersByQuery [ queryKey ] = listeners ; } listeners . push ( listener ) ; } private removeListener ( listener ) { const listenersByQuery = this . listenersByTypeAndQuery [ listener . inverse . appliedToType ] ; if ( listenersByQuery ) { const queryKey = listener . inverse . affected . toDescriptiveString ( ) ; const listeners = listenersByQuery [ queryKey ] ; if ( listeners ) { const index = listeners . indexOf ( listener ) ; if ( index >= 0 ) { listeners . splice ( index , 1 ) ; } } } } private async notifyFactSaved ( fact ) { const listenersByQuery = this . listenersByTypeAndQuery [ fact . type ] ; if ( listenersByQuery ) { for ( const queryKey in listenersByQuery ) { const listeners = listenersByQuery [ queryKey ] ; if ( listeners && listeners . length > 0 ) { const query = listeners [ 0 ] . inverse . affected ; const affected = await this . inner . query ( fact , query ) ; await mapAsync ( listeners , async ( listener ) => { const matching = affected . filter ( path => { const last = path [ path . length - 1 ] ; return last . hash === listener . match . hash && last . type === listener . match . type ; } ) ; await mapAsync ( matching , async ( backtrack ) => { await this . notifyListener ( [ { type : fact . type , hash : fact . hash } ] . concat ( backtrack ) . reverse ( ) . slice ( 1 ) , listener ) ; } ) ; } ) ; } } } } private async notifyListener ( prefix , listener ) { const fact = prefix [ prefix . length - 1 ] ; if ( listener . inverse . added && listener . added ) { const added = await this . inner . query ( fact , listener . inverse . added ) ; if ( added . length > 0 ) { const paths = added . map ( path => prefix . concat ( path ) ) ; listener . added ( paths ) ; } } if ( listener . inverse . removed && listener . removed ) { const removed = prefix . slice ( 0 , prefix . length - listener . inverse . removed . getPathLength ( ) ) ; if ( removed . length > 0 ) { listener . removed ( [ removed ] ) ; } } } }