feat: fix filters, add logging (#29)

- filter functions are now dynamic
- fix types for subscribe options
- payloads need to be Record<string, any>
- logging is now in a lot of places, you can add your own if you like, it should act like npm's `debug`
- you don't have to have payloads for complete messages

Author: reconbot

lib/gateway.ts

@@ -14,13 +14,15 @@ import { pong } from './messages/pong'
 
 export const handleGatewayEvent = (server: ServerClosure): ApiGatewayHandler<APIGatewayWebSocketEvent, WebsocketResponse> => async (event) => {
   if (!event.requestContext) {
+    server.log('handleGatewayEvent unknown')
     return {
       statusCode: 200,
       body: '',
     }
   }
 
   if (event.requestContext.eventType === 'CONNECT') {
+    server.log('handleGatewayEvent CONNECT', { connectionId: event.requestContext.connectionId })
     await server.onConnect?.({ event })
     return {
       statusCode: 200,
@@ -33,6 +35,7 @@ export const handleGatewayEvent = (server: ServerClosure): ApiGatewayHandler<API
 
   if (event.requestContext.eventType === 'MESSAGE') {
     const message = event.body === null ? null : JSON.parse(event.body)
+    server.log('handleGatewayEvent MESSAGE', { connectionId: event.requestContext.connectionId, type: message.type })
 
     if (message.type === MessageType.ConnectionInit) {
       await connection_init({ server, event, message })
@@ -76,13 +79,15 @@ export const handleGatewayEvent = (server: ServerClosure): ApiGatewayHandler<API
   }
 
   if (event.requestContext.eventType === 'DISCONNECT') {
+    server.log('handleGatewayEvent DISCONNECT', { connectionId: event.requestContext.connectionId })
     await disconnect({ server, event, message: null })
     return {
       statusCode: 200,
       body: '',
     }
   }
 
+  server.log('handleGatewayEvent UNKNOWN', { connectionId: event.requestContext.connectionId })
   return {
     statusCode: 200,
     body: '',

lib/index-test.ts

@@ -12,7 +12,7 @@ describe('createInstance', () => {
     })
 
     after(async () => {
-      tables.end()
+      await tables.end()
     })
 
     it('is type compatible with aws-lambda handler', async () => {

lib/makeServerClosure.ts

@@ -3,9 +3,11 @@ import { ServerArgs, ServerClosure } from './types'
 import { createModel } from './model/createModel'
 import { Subscription } from './model/Subscription'
 import { Connection } from './model/Connection'
+import { log } from './utils/logger'
 
 export function makeServerClosure(opts: ServerArgs): ServerClosure {
   return {
+    log: log,
     ...opts,
     model: {
       Subscription: createModel({

lib/messages/complete.ts

@@ -3,7 +3,7 @@ import { parse } from 'graphql'
 import { CompleteMessage } from 'graphql-ws'
 import { buildExecutionContext } from 'graphql/execution/execute'
 import { collect } from 'streaming-iterables'
-import { SubscribePseudoIterable, MessageHandler } from '../types'
+import { SubscribePseudoIterable, MessageHandler, PubSubEvent } from '../types'
 import { deleteConnection } from '../utils/deleteConnection'
 import { constructContext } from '../utils/constructContext'
 import { getResolverAndArgs } from '../utils/getResolverAndArgs'
@@ -37,7 +37,7 @@ export const complete: MessageHandler<CompleteMessage> =
 
       const [field, root, args, context, info] = getResolverAndArgs(server)(execContext)
 
-      const onComplete = (field?.subscribe as SubscribePseudoIterable)?.onComplete
+      const onComplete = (field?.subscribe as SubscribePseudoIterable<PubSubEvent>)?.onComplete
       if (onComplete) {
         await onComplete(root, args, context, info)
       }

lib/messages/disconnect.ts

@@ -4,7 +4,7 @@ import { equals } from '@aws/dynamodb-expressions'
 import { buildExecutionContext } from 'graphql/execution/execute'
 import { constructContext } from '../utils/constructContext'
 import { getResolverAndArgs } from '../utils/getResolverAndArgs'
-import { SubscribePseudoIterable, MessageHandler } from '../types'
+import { SubscribePseudoIterable, MessageHandler, PubSubEvent } from '../types'
 import { isArray } from '../utils/isArray'
 import { collect } from 'streaming-iterables'
 import { Connection } from '../model/Connection'
@@ -49,7 +49,7 @@ export const disconnect: MessageHandler<null> =
 
               const [field, root, args, context, info] = getResolverAndArgs(server)(execContext)
 
-              const onComplete = (field?.subscribe as SubscribePseudoIterable)?.onComplete
+              const onComplete = (field?.subscribe as SubscribePseudoIterable<PubSubEvent>)?.onComplete
               if (onComplete) {
                 await onComplete(root, args, context, info)
               }

lib/messages/subscribe-test.ts

@@ -19,7 +19,7 @@ describe('messages/subscribe', () => {
   })
 
   afterEach(async () => {
-    tables.end()
+    await tables.end()
   })
 
   it('executes a query/mutation', async () => {

lib/messages/subscribe.ts

@@ -6,7 +6,7 @@ import {
   assertValidExecutionArguments,
   execute,
 } from 'graphql/execution/execute'
-import { APIGatewayWebSocketEvent, ServerClosure, SubscribeHandler, MessageHandler } from '../types'
+import { APIGatewayWebSocketEvent, ServerClosure, MessageHandler, SubscribePseudoIterable, PubSubEvent } from '../types'
 import { constructContext } from '../utils/constructContext'
 import { getResolverAndArgs } from '../utils/getResolverAndArgs'
 import { sendMessage } from '../utils/sendMessage'
@@ -25,9 +25,11 @@ export const subscribe: MessageHandler<SubscribeMessage> =
   }
 
 const setupSubscription: MessageHandler<SubscribeMessage> = async ({ server, event, message }) => {
+  const connectionId = event.requestContext.connectionId
+
   const connection = await server.mapper.get(
     Object.assign(new server.model.Connection(), {
-      id: event.requestContext.connectionId,
+      id: connectionId,
     }),
   )
   const connectionParams = connection.payload || {}
@@ -39,7 +41,7 @@ const setupSubscription: MessageHandler<SubscribeMessage> = async ({ server, eve
     throw new AggregateError(errors)
   }
 
-  const contextValue = await constructContext({ server, connectionParams, connectionId: connection.id })
+  const contextValue = await constructContext({ server, connectionParams, connectionId })
 
   const execContext = buildExecutionContext(
     server.schema,
@@ -74,33 +76,33 @@ const setupSubscription: MessageHandler<SubscribeMessage> = async ({ server, eve
     throw new Error('No field')
   }
 
-  const { topicDefinitions, onSubscribe, onAfterSubscribe } = await (field.subscribe as SubscribeHandler)(
-    root,
-    args,
-    context,
-    info,
-  )
+  const { topicDefinitions, onSubscribe, onAfterSubscribe } = field.subscribe as SubscribePseudoIterable<PubSubEvent>
 
+  server.log('onSubscribe', { onSubscribe: !!onSubscribe })
   await onSubscribe?.(root, args, context, info)
 
   await Promise.all(topicDefinitions.map(async ({ topic, filter }) => {
+    const filterData = typeof filter === 'function' ? await filter(root, args, context, info) : filter
+
     const subscription = Object.assign(new server.model.Subscription(), {
-      id: `${event.requestContext.connectionId}|${message.id}`,
+      id: `${connectionId}|${message.id}`,
       topic,
-      filter: filter || {},
+      filter: filterData || {},
       subscriptionId: message.id,
       subscription: {
         variableValues: args,
         ...message.payload,
       },
-      connectionId: event.requestContext.connectionId,
+      connectionId,
       connectionParams,
       requestContext: event.requestContext,
       ttl: connection.ttl,
     })
+    server.log('subscribe:putSubscription %j', subscription)
     await server.mapper.put(subscription)
   }))
 
+  server.log('onAfterSubscribe', { onAfterSubscribe: !!onAfterSubscribe })
   await onAfterSubscribe?.(root, args, context, info)
 }
 
@@ -125,6 +127,8 @@ const validateMessage = (server: ServerClosure) => (message: SubscribeMessage) =
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 async function executeQuery(server: ServerClosure, message: SubscribeMessage, contextValue: any, event: APIGatewayWebSocketEvent) {
+  server.log('executeQuery', { connectionId: event.requestContext.connectionId })
+
   const result = await execute(
     server.schema,
     parse(message.payload.query),

lib/model/createModel.ts

@@ -1,7 +1,7 @@
 import { DynamoDbTable } from '@aws/dynamodb-data-mapper'
-import { Class } from '../types'
 
-export const createModel = <T extends Class>({
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export const createModel = <T extends { new(...args: any[]): any }>({
   model,
   table,
 }: {

lib/pubsub/complete-test.ts

@@ -0,0 +1,18 @@
+import { tables } from '@architect/sandbox'
+import { mockServerContext } from '../test/mockServer'
+import { complete } from './complete'
+
+describe('pubsub:complete', () => {
+  before(async () => {
+    await tables.start({ cwd: './mocks/arc-basic-events', quiet: true })
+  })
+
+  after(async () => {
+    await tables.end()
+  })
+
+  it('takes a topic', async () => {
+    const server = await mockServerContext()
+    await complete(server)({ topic: 'Topic12' })
+  })
+})

lib/pubsub/complete.ts

@@ -2,15 +2,17 @@ import AggregateError from 'aggregate-error'
 import { parse } from 'graphql'
 import { CompleteMessage, MessageType } from 'graphql-ws'
 import { buildExecutionContext } from 'graphql/execution/execute'
-import { ServerClosure, PubSubEvent, SubscribePseudoIterable } from '../types'
+import { ServerClosure, PubSubEvent, SubscribePseudoIterable, PartialBy } from '../types'
 import { sendMessage } from '../utils/sendMessage'
 import { constructContext } from '../utils/constructContext'
 import { getResolverAndArgs } from '../utils/getResolverAndArgs'
 import { isArray } from '../utils/isArray'
 import { getFilteredSubs } from './getFilteredSubs'
 
-export const complete = (server: ServerClosure) => async (event: PubSubEvent): Promise<void> => {
+export const complete = (server: ServerClosure) => async (event: PartialBy<PubSubEvent, 'payload'>): Promise<void> => {
   const subscriptions = await getFilteredSubs({ server, event })
+  server.log('pubsub:complete %j', { event, subscriptions })
+
   const iters = subscriptions.map(async (sub) => {
     const message: CompleteMessage = {
       id: sub.subscriptionId,
@@ -38,7 +40,7 @@ export const complete = (server: ServerClosure) => async (event: PubSubEvent): P
 
     const [field, root, args, context, info] = getResolverAndArgs(server)(execContext)
 
-    const onComplete = (field?.subscribe as SubscribePseudoIterable)?.onComplete
+    const onComplete = (field?.subscribe as SubscribePseudoIterable<PubSubEvent>)?.onComplete
     if (onComplete) {
       await onComplete(root, args, context, info)
     }