import { FieldNode , GraphQLResolveInfo } from 'graphql' ; import { ddl , asyncIteratorQueryStream } from '@confluentinc/ksqldb-client' ; import { runCommand } from './requester' ; import { Resolver , SubscriptionResolver , ResolverFields , KsqlDBGraphResolver , } from './type/definition' ; const getNameFromResolveInfo = ( info ) : string | null => { const node = info . fieldNodes [ 0 ] ; if ( ! node || ! node . selectionSet ) { return null ; } return escape ( node . name . value ) ; } ; const createResolver = ( resolver ) => ( resolvers , key ) : Resolver => { resolvers [ key ] = resolver ; return resolvers ; } ; export const createInsertStatement = ( info , args ) : string | null => { const node = info . fieldNodes [ 0 ] ; if ( ! node || ! node . selectionSet ) { return null ; } const name = getNameFromResolveInfo ( info ) ; const keys = Object . keys ( args ) . map ( key => escape ( key ) ) ; const values = Object . values ( args ) . map ( val => typeof val === 'string' ? `'${ escape ( val ) }'` : val ) ; const command = `INSERT INTO ${ name } (${ keys . join ( ', ' ) }) VALUES (${ values . join ( ', ' ) });` ; return command ; } ; export class ResolverGenerator { whitelistFields : Set < string > = new Set ( ) ; queryResolvers = { } ; mutationResolvers = { } ; subscriptionResolvers = { } ; constructor ( protected fields ) { const { queryFields , subscriptionFields , mutationFields } = fields ; const queries = Object . keys ( queryFields ) ; const mutations = Object . keys ( mutationFields ) ; const subscriptions = Object . keys ( subscriptionFields ) ; queries . concat ( subscriptions , mutations ) . reduce ( ( accum , key ) => { let args ; if ( queryFields [ key ] ) { args = queryFields [ key ] . args ; } else if ( subscriptionFields [ key ] ) { args = subscriptionFields [ key ] . args ; } else if ( mutationFields [ key ] ) { args = mutationFields [ key ] . args ; } if ( args ) { Object . keys ( args ) . forEach ( ( key ) => { accum . add ( key ) ; } ) ; } return accum ; } , this . whitelistFields ) ; this . queryResolvers = queries . reduce ( createResolver ( this . handleQueryResolve ) , { } ) ; this . mutationResolvers = mutations . reduce ( createResolver ( this . handleMutationResolve ) , { } ) ; this . subscriptionResolvers = subscriptions . reduce ( ( resolvers , key ) => { resolvers [ key ] = { subscribe : this . handleSubscriptionResolve , } ; return resolvers ; } , { } ) ; } getFields ( selections ) { const fields = selections . filter ( ( { name } ) => this . whitelistFields . has ( name . value ) ) . map ( ( { name } ) => { return escape ( name . value ) ; } ) ; if ( fields . length === 0 ) { throw new Error ( 'No ksqlDB fields available.' ) ; } return fields ; } generateStatement = ( info , args ) : string | null => { const node = info . fieldNodes [ 0 ] ; if ( ! node || ! node . selectionSet ) { return null ; } const command = [ ] ; const selections = node . selectionSet . selections as Array < FieldNode > ; const fields = this . getFields ( selections ) ; const name = getNameFromResolveInfo ( info ) ; command . push ( `select ${ fields . join ( ', ' ) } from ${ name }` ) ; const whereArguments = Object . keys ( args ) ; if ( whereArguments . length > 0 ) { command . push ( `where ${ whereArguments . map ( key => { return `${ key }='${ escape ( args [ key ] ) }'` ; } ) . join ( ' and ' ) }` ) ; } return `${ command . join ( ' ' ) }` ; } ; handleQueryResolve : KsqlDBGraphResolver = async ( obj , args , { ksqlDB } , info ) : Promise < any > => { const sql = `${ this . generateStatement ( info , args ) };` ; if ( ! sql ) { return null ; } const stream = asyncIteratorQueryStream ( ksqlDB . session , { sql } , 'query' ) ; for await ( const row of stream ) { const { query } = row ; return Object . keys ( query ) . reduce ( ( accum , qry ) => { if ( query [ qry ] . done ) { const { value } = query [ qry ] ; try { const parsedVal = JSON . parse ( value [ 0 ] ) ; accum [ qry ] = parsedVal [ 0 ] ; } catch ( e ) { console . error ( e ) ; } } else { accum [ qry ] = query [ qry ] ; } return accum ; } , { } ) ; } } ; handleSubscriptionResolve : KsqlDBGraphResolver = async ( obj , args , { ksqlDB } , info ) => { const sql = `${ this . generateStatement ( info , args ) } emit changes;` ; if ( ! sql ) { return Promise . resolve ( new Error ( 'Unable to generate ksqlDB from graphql statement.' ) ) ; } const nameKey = getNameFromResolveInfo ( info ) as string ; const stream = asyncIteratorQueryStream ( ksqlDB . session , { sql } , nameKey ) ; return stream ; } ; handleMutationResolve : KsqlDBGraphResolver = async ( obj , args , { ksqlDB } , info ) => { const { options , session } = ksqlDB ; const command = createInsertStatement ( info , args ) ; try { if ( ! command ) { throw new Error ( 'Unable to create insert statement from graphql args' ) ; } const response = await ddl ( session , { ksql : command } ) ; if ( response && response . error ) { console . warn ( 'new api unavailable, falling back to default REST.' ) ; const response = await runCommand ( command , options ) ; return { command , statusCode : response . statusCode } ; } return { command , statusCode : response && response . statusCode } ; } catch ( e ) { if ( e . response ) { throw new Error ( e . response . data . message ) ; } throw new Error ( e ) ; } } ; }