import { BoosterConfig , Logger , Instance , ReadModelEnvelope , ReadModelInterface , SubscriptionEnvelope , GraphQLData , } from '@boostercloud/framework-types' ; import { GraphQLSchema , DocumentNode } from 'graphql' ; import * as graphql from 'graphql' ; import { GraphQLGenerator } from './services/graphql/graphql-generator' ; import { FilteredReadModelPubSub , ReadModelPubSub } from './services/pub-sub/read-model-pub-sub' ; import { GraphQLResolverContext } from './services/graphql/common' ; import { ExecutionResult } from 'graphql/execution/execute' ; import { Promises } from '@boostercloud/framework-common-helpers' ; export class BoosterSubscribersNotifier { private readonly graphQLSchema : GraphQLSchema ; public constructor ( private config , private logger ) { this . graphQLSchema = GraphQLGenerator . generateSchema ( config , logger ) ; } public async dispatch ( request ) { try { this . logger . debug ( 'Received the following event for subscription dispatching: ' , request ) ; const readModelEnvelopes = await this . config . provider . readModels . rawToEnvelopes ( this . config , this . logger , request ) ; this . logger . debug ( '[SubsciptionDispatcher] The following ReadModels were updated: ' , readModelEnvelopes ) ; const subscriptions = await this . getSubscriptions ( readModelEnvelopes ) ; this . logger . debug ( '[SubsciptionDispatcher] Found the following subscriptions for those read models: ' , subscriptions ) ; const pubSub = this . getPubSub ( readModelEnvelopes ) ; await Promises . allSettledAndFulfilled ( subscriptions . map ( this . runSubscriptionAndNotify . bind ( this , pubSub ) ) ) ; } catch ( e ) { this . logger . error ( e ) ; } } private getReadModelInstance ( envelope ) { const readModelMetadata = this . config . readModels [ envelope . typeName ] ; if ( ! readModelMetadata ) { throw new Error ( 'Could not get information about read model with name: ' + envelope . typeName ) ; } const readModelInstance = new readModelMetadata . class ( ) ; Object . assign ( readModelInstance , envelope . value ) ; return readModelInstance ; } private async getSubscriptions ( readModelEnvelopes ) { const readModelNames = readModelEnvelopes . map ( ( readModelEnvelope ) => readModelEnvelope . typeName ) ; const readModelUniqueNames = [ ... new Set ( readModelNames ) ] ; const subscriptionSets = await Promise . all ( readModelUniqueNames . map ( ( name ) => this . config . provider . readModels . fetchSubscriptions ( this . config , this . logger , name ) ) ) ; return subscriptionSets . flat ( ) ; } private getPubSub ( readModelEnvelopes ) { const readModelInstances = readModelEnvelopes . map ( this . getReadModelInstance , this ) ; return new FilteredReadModelPubSub ( readModelInstances ) ; } private async runSubscriptionAndNotify ( pubSub , subscription ) { const context = { connectionID : subscription . connectionID , requestID : subscription . requestID , user : subscription . currentUser , operation : subscription . operation , pubSub , storeSubscriptions : false , } ; const document = this . parseSubscriptionQuery ( subscription . operation . query ) ; this . logger . debug ( 'Running subscription with context: ' , context ) ; const iterator = await graphql . subscribe ( { contextValue : context , document : document , schema : this . graphQLSchema , variableValues : subscription . operation . variables , } ) ; if ( 'next' in iterator ) { return this . processSubscriptionsIterator ( iterator , subscription ) ; } throw iterator . errors ; } private parseSubscriptionQuery ( query ) { const document = graphql . parse ( query ) ; const errors = graphql . validate ( this . graphQLSchema , document ) ; if ( errors . length > 0 ) { throw errors ; } return document ; } private async processSubscriptionsIterator ( iterator , subscription ) { const notificationPromises = [ ] ; for await ( const result of iterator ) { notificationPromises . push ( this . notifyWithGraphQLResult ( subscription , result ) ) ; } return Promise . all ( notificationPromises ) ; } private async notifyWithGraphQLResult ( subscription , result ) { if ( result . errors ) { throw result . errors ; } const readModel = result . data as ReadModelInterface ; const message = new GraphQLData ( subscription . operation . id ! , { data : readModel } ) ; this . logger . debug ( `Notifying connectionID '${ subscription . connectionID }' with the following wrappeed read model: ` , readModel ) ; await this . config . provider . connections . sendMessage ( this . config , subscription . connectionID , message ) ; this . logger . debug ( 'Notifications sent' ) ; } }