import { GraphQLSchema , GraphQLString , GraphQLFloat , GraphQLList , GraphQLObjectType , printSchema , isInputType , GraphQLScalarType , GraphQLObjectTypeConfig , } from 'graphql' ; import { RequestOptions } from 'http' ; import { ResolverGenerator } from './resolvers' ; import { Config , Field , KsqlDBResponse , KSqlDBEntities , ResolverFields } from './type/definition' ; import { Missing , KsqlDBMutation } from './graphQLObjectTypes' ; import { runCommand } from './requester' ; const TypeMap = { STRING : GraphQLString , VARCHAR : GraphQLString , BIGINT : GraphQLFloat , DOUBLE : GraphQLFloat , INTEGER : GraphQLFloat , ARRAY : { STRING : new GraphQLList ( GraphQLString ) , VARCHAR : new GraphQLList ( GraphQLString ) , BIGINT : new GraphQLList ( GraphQLFloat ) , INTEGER : new GraphQLList ( GraphQLFloat ) , DOUBLE : new GraphQLList ( GraphQLFloat ) , } , STRUCT : { } , } ; const setSchemaType = ( accum , field ) : void => { if ( TypeMap [ field . schema . type ] == null ) { console . error ( `type ${ field . schema . type } is not supported` ) ; return ; } if ( field . schema . memberSchema ?. type != null ) { const sclarType = TypeMap [ field . schema . type ] [ field . schema . memberSchema . type ] as GraphQLScalarType ; accum [ field . name ] = { type : sclarType , } ; } else { const sclarType = TypeMap [ field . schema . type ] as GraphQLScalarType ; accum [ field . name ] = { type : sclarType , } ; } } ; const buildSchemaObject = ( accum , field ) : KSqlDBEntities => { if ( field . schema . fields == null ) { setSchemaType ( accum , field ) ; } else if ( Array . isArray ( field . schema . fields ) ) { const fields = field . schema . fields . reduce ( buildSchemaObject , { } ) ; if ( accum [ field . name ] == null ) { accum [ field . name ] = { type : new GraphQLObjectType ( { name : field . name , fields : fields } ) } ; } else { console . warn ( `${ field . name } already exists.` ) ; } } return accum ; } ; export const generateSchemaFromKsql = ( { name , fields , } ) : GraphQLObjectTypeConfig < void , void > => { const schemaFields = fields . reduce ( buildSchemaObject , { } ) ; return { name , fields : schemaFields , } ; } ; const generateGraqphQLArgs = ( fields ) : any => Object . keys ( fields ) . reduce ( ( accum , filter ) => { if ( isInputType ( fields [ filter ] . type ) ) { accum [ filter ] = fields [ filter ] ; } return accum ; } , { } ) ; function generateQueries ( streams , subscriptionFields ) { return ( accum , query ) : any => { const schemaType = new GraphQLObjectType ( query ) ; const ksqlDBQuery = streams . find ( stream => stream . name === query . name ) ; if ( ksqlDBQuery != null && ksqlDBQuery . writeQueries . length > 0 ) { const args = generateGraqphQLArgs ( query . fields ) ; if ( subscriptionFields [ query . name ] != null ) { accum [ query . name ] = subscriptionFields [ query . name ] ; } else { accum [ query . name ] = { type : schemaType , args , } ; } } return accum ; } ; } function generateSubscription ( accum , query ) { const schemaType = new GraphQLObjectType ( query ) ; const args = generateGraqphQLArgs ( query . fields ) ; accum [ query . name ] = { type : schemaType , args , } ; return accum ; } function generateMutations ( accum , query ) { const args = generateGraqphQLArgs ( query . fields ) ; accum [ query . name ] = { type : KsqlDBMutation , args , } ; return accum ; } export const generateSchemaAndFields = ( streams ) : { schema : GraphQLSchema ; fields : ResolverFields ; } => { const schemas = [ ] ; for ( const stream of streams ) { schemas . push ( generateSchemaFromKsql ( stream ) ) ; } const subscriptionFields = schemas . reduce ( generateSubscription , { } ) ; const mutationFields = schemas . reduce ( generateMutations , { } ) ; let queryFields = schemas . reduce ( generateQueries ( streams , subscriptionFields ) , { } ) ; if ( Object . keys ( queryFields ) . length === 0 ) { console . error ( 'No materalized views have been registered.' , 'Only subscriptions and mutations will be work properly.' , 'Defaulting `type Query` to null scalar since it is required by graphQL.' ) ; queryFields = { KsqlDBGraphQLError : Missing } ; } return { schema : new GraphQLSchema ( { query : new GraphQLObjectType ( { name : 'Query' , fields : queryFields , } ) , subscription : new GraphQLObjectType ( { name : 'Subscription' , fields : subscriptionFields , } ) , mutation : new GraphQLObjectType ( { name : 'Mutation' , fields : mutationFields } ) , } ) , fields : { queryFields : Object . keys ( queryFields ) . filter ( key => { return queryFields [ key ] !== Missing ; } ) . reduce ( ( accum , key ) => { accum [ key ] = queryFields [ key ] ; return accum ; } , { } ) , subscriptionFields , mutationFields , } , } ; } ; const schemas = async ( options ) : Promise < { schema : GraphQLSchema ; fields : ResolverFields ; } | undefined > => { const ksql = 'show streams extended; show tables extended;' ; try { const response = await runCommand ( ksql , options ) ; const streams = response . data [ 0 ] . sourceDescriptions ; const tables = response . data [ 1 ] . sourceDescriptions ; if ( streams . length === 0 ) { throw new Error ( `No ksql tables exist on ksql server ${ options . hostname }:${ options . port }` ) ; } return generateSchemaAndFields ( streams . concat ( tables ) ) ; } catch ( e ) { console . error ( `Could not generate schemas:` , e . message ) ; } } ; export function buildKsqlDBGraphQL ( { options , } ) { return new Promise ( resolve => { ( async function run ( ) : Promise < void > { try { const result = await schemas ( options ) ; if ( result ) { console . log ( printSchema ( result . schema ) ) ; const { queryResolvers , subscriptionResolvers , mutationResolvers , } = new ResolverGenerator ( result . fields ) ; resolve ( { schemas : result . schema , queryResolvers , subscriptionResolvers , mutationResolvers , } ) ; } else { throw new Error ( 'Unable to create schemas and resolvers' ) ; } } catch ( e ) { throw new Error ( e ) ; } } ) ( ) ; } ) ; }