feat: hook into published subscribe values

.changeset/fifty-kids-boil.md

@@ -0,0 +1,6 @@
+---
+'@envelop/core': minor
+'@envelop/types': minor
+---
+
+allow hooking into published subscribe values

.changeset/six-crabs-flash.md

@@ -0,0 +1,5 @@
+---
+'@envelop/execute-subscription-event': patch
+---
+
+initial release

benchmark/k6.js

@@ -50,7 +50,7 @@ export const options = buildOptions({
   'graphql-js': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=28'],
+    http_req_duration: ['p(95)<=35'],
     graphql_execute: ['p(95)<=2'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
@@ -61,7 +61,7 @@ export const options = buildOptions({
   'envelop-just-cache': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=20'],
+    http_req_duration: ['p(95)<=23'],
     graphql_execute: ['p(95)<=1'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
@@ -72,23 +72,23 @@ export const options = buildOptions({
   'prom-tracing': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=45'],
-    graphql_execute: ['p(95)<=4'],
+    http_req_duration: ['p(95)<=52'],
+    graphql_execute: ['p(95)<=6'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
     graphql_parse: ['p(95)<=1'],
     envelop_init: ['p(95)<=1'],
-    envelop_total: ['p(95)<=4'],
+    envelop_total: ['p(95)<=6'],
   },
   'envelop-cache-and-no-internal-tracing': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=15'],
+    http_req_duration: ['p(95)<=18'],
   },
   'envelop-cache-jit': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=14'],
+    http_req_duration: ['p(95)<=16'],
     graphql_execute: ['p(95)<=1'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],

packages/core/src/graphql-typings.d.ts

@@ -1,4 +1,4 @@
 declare module 'graphql/jsutils/isAsyncIterable' {
-  function isAsyncIterable(input: unknown): input is AsyncIterable<any>;
+  function isAsyncIterable(input: unknown): input is AsyncIterableIterator<any>;
   export default isAsyncIterable;
 }

packages/core/src/orchestrator.ts

@@ -16,25 +16,30 @@ import {
   SubscribeResultHook,
   TypedSubscriptionArgs,
   TypedExecutionArgs,
+  SubscribeFunction,
+  OnSubscribeResultResultOnNextHook,
+  OnSubscribeResultResultOnEndHook,
+  OnExecuteDoneHookResultOnNextHook,
+  OnExecuteDoneHookResultOnEndHook,
+  ExecuteFunction,
+  AsyncIterableIteratorOrValue,
 } from '@envelop/types';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 import {
   DocumentNode,
   execute,
-  ExecutionArgs,
   ExecutionResult,
   GraphQLError,
-  GraphQLFieldResolver,
   GraphQLSchema,
-  GraphQLTypeResolver,
   parse,
   specifiedRules,
   subscribe,
-  SubscriptionArgs,
   validate,
   ValidationRule,
 } from 'graphql';
 import { Maybe } from 'graphql/jsutils/Maybe';
 import { prepareTracedSchema, resolversHooksSymbol } from './traced-schema';
+import { finalAsyncIterator, makeExecute, makeSubscribe, mapAsyncIterator } from './utils';
 
 export type EnvelopOrchestrator<
   InitialContext extends ArbitraryObject = ArbitraryObject,
@@ -272,32 +277,9 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
       }
     : initialContext => orchestratorCtx => orchestratorCtx ? { ...initialContext, ...orchestratorCtx } : initialContext;
 
-  const customSubscribe = async (
-    argsOrSchema: SubscriptionArgs | GraphQLSchema,
-    document?: DocumentNode,
-    rootValue?: any,
-    contextValue?: any,
-    variableValues?: Maybe<{ [key: string]: any }>,
-    operationName?: Maybe<string>,
-    fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
-    subscribeFieldResolver?: Maybe<GraphQLFieldResolver<any, any>>
-  ) => {
-    const args: SubscriptionArgs =
-      argsOrSchema instanceof GraphQLSchema
-        ? {
-            schema: argsOrSchema,
-            document: document!,
-            rootValue,
-            contextValue,
-            variableValues,
-            operationName,
-            fieldResolver,
-            subscribeFieldResolver,
-          }
-        : argsOrSchema;
-
+  const customSubscribe = makeSubscribe(async args => {
     const onResolversHandlers: OnResolverCalledHook[] = [];
-    let subscribeFn: typeof subscribe = subscribe;
+    let subscribeFn = subscribe as SubscribeFunction;
 
     const afterCalls: SubscribeResultHook[] = [];
     let context = args.contextValue || {};
@@ -333,46 +315,49 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
       contextValue: context,
     });
 
+    const onNextHandler: OnSubscribeResultResultOnNextHook[] = [];
+    const onEndHandler: OnSubscribeResultResultOnEndHook[] = [];
+
     for (const afterCb of afterCalls) {
-      afterCb({
+      const hookResult = afterCb({
         result,
         setResult: newResult => {
           result = newResult;
         },
       });
+      if (hookResult) {
+        if (hookResult.onNext) {
+          onNextHandler.push(hookResult.onNext);
+        }
+        if (hookResult.onEnd) {
+          onEndHandler.push(hookResult.onEnd);
+        }
+      }
     }
 
+    if (onNextHandler.length && isAsyncIterable(result)) {
+      result = mapAsyncIterator(result, async result => {
+        for (const onNext of onNextHandler) {
+          await onNext({ result, setResult: newResult => (result = newResult) });
+        }
+        return result;
+      });
+    }
+    if (onEndHandler.length && isAsyncIterable(result)) {
+      result = finalAsyncIterator(result, () => {
+        for (const onEnd of onEndHandler) {
+          onEnd();
+        }
+      });
+    }
     return result;
-  };
+  });
 
   const customExecute = beforeCallbacks.execute.length
-    ? async (
-        argsOrSchema: ExecutionArgs | GraphQLSchema,
-        document?: DocumentNode,
-        rootValue?: any,
-        contextValue?: any,
-        variableValues?: Maybe<{ [key: string]: any }>,
-        operationName?: Maybe<string>,
-        fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
-        typeResolver?: Maybe<GraphQLTypeResolver<any, any>>
-      ) => {
-        const args: ExecutionArgs =
-          argsOrSchema instanceof GraphQLSchema
-            ? {
-                schema: argsOrSchema,
-                document: document!,
-                rootValue,
-                contextValue,
-                variableValues,
-                operationName,
-                fieldResolver,
-                typeResolver,
-              }
-            : argsOrSchema;
-
+    ? makeExecute(async args => {
         const onResolversHandlers: OnResolverCalledHook[] = [];
-        let executeFn: typeof execute = execute;
-        let result: ExecutionResult;
+        let executeFn = execute as ExecuteFunction;
+        let result: AsyncIterableIteratorOrValue<ExecutionResult>;
 
         const afterCalls: OnExecuteDoneHook[] = [];
         let context = args.contextValue || {};
@@ -429,17 +414,44 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
           contextValue: context,
         });
 
+        const onNextHandler: OnExecuteDoneHookResultOnNextHook[] = [];
+        const onEndHandler: OnExecuteDoneHookResultOnEndHook[] = [];
+
         for (const afterCb of afterCalls) {
-          afterCb({
+          const hookResult = afterCb({
             result,
             setResult: newResult => {
               result = newResult;
             },
           });
+          if (hookResult) {
+            if (hookResult.onNext) {
+              onNextHandler.push(hookResult.onNext);
+            }
+            if (hookResult.onEnd) {
+              onEndHandler.push(hookResult.onEnd);
+            }
+          }
+        }
+
+        if (onNextHandler.length && isAsyncIterable(result)) {
+          result = mapAsyncIterator(result, async result => {
+            for (const onNext of onNextHandler) {
+              await onNext({ result, setResult: newResult => (result = newResult) });
+            }
+            return result;
+          });
+        }
+        if (onEndHandler.length && isAsyncIterable(result)) {
+          result = finalAsyncIterator(result, () => {
+            for (const onEnd of onEndHandler) {
+              onEnd();
+            }
+          });
         }
 
         return result;
-      }
+      })
     : execute;
 
   initDone = true;
@@ -463,7 +475,7 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
     init,
     parse: customParse,
     validate: customValidate,
-    execute: customExecute,
+    execute: customExecute as ExecuteFunction,
     subscribe: customSubscribe,
     contextFactory: customContextFactory,
   };

packages/core/src/plugins/use-error-handler.ts

@@ -1,13 +1,31 @@
 import { Plugin } from '@envelop/types';
-import { GraphQLError } from 'graphql';
+import { ExecutionResult, GraphQLError } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
-export const useErrorHandler = (errorHandler: (errors: readonly GraphQLError[]) => void): Plugin => ({
+export type ErrorHandler = (errors: readonly GraphQLError[]) => void;
+
+const makeHandleResult =
+  (errorHandler: ErrorHandler) =>
+  ({ result }: { result: ExecutionResult }) => {
+    if (result.errors?.length) {
+      errorHandler(result.errors);
+    }
+  };
+
+export const useErrorHandler = (errorHandler: ErrorHandler): Plugin => ({
   onExecute() {
+    const handleResult = makeHandleResult(errorHandler);
     return {
-      onExecuteDone: ({ result }) => {
-        if (result.errors?.length) {
-          errorHandler(result.errors);
+      onExecuteDone({ result }) {
+        if (isAsyncIterable(result)) {
+          return {
+            onNext({ result }) {
+              handleResult({ result });
+            },
+          };
         }
+        handleResult({ result });
+        return undefined;
       },
     };
   },

packages/core/src/plugins/use-masked-errors.ts

@@ -1,5 +1,6 @@
 import { Plugin } from '@envelop/types';
-import { GraphQLError } from 'graphql';
+import { ExecutionResult, GraphQLError } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 export class EnvelopError extends GraphQLError {
   constructor(message: string, extensions?: Record<string, any>) {
@@ -21,16 +22,31 @@ export type UseMaskedErrorsOpts = {
   formatError?: FormatErrorHandler;
 };
 
+const makeHandleResult =
+  (format: FormatErrorHandler) =>
+  ({ result, setResult }: { result: ExecutionResult; setResult: (result: ExecutionResult) => void }) => {
+    if (result.errors != null) {
+      setResult({ ...result, errors: result.errors.map(error => format(error)) });
+    }
+  };
+
 export const useMaskedErrors = (opts?: UseMaskedErrorsOpts): Plugin => {
   const format = opts?.formatError ?? formatError;
+  const handleResult = makeHandleResult(format);
 
   return {
-    onExecute: () => {
+    onExecute() {
       return {
-        onExecuteDone: ({ result, setResult }) => {
-          if (result.errors != null) {
-            setResult({ ...result, errors: result.errors.map(error => format(error)) });
+        onExecuteDone({ result, setResult }) {
+          if (isAsyncIterable(result)) {
+            return {
+              onNext: ({ result, setResult }) => {
+                handleResult({ result, setResult });
+              },
+            };
           }
+          handleResult({ result, setResult });
+          return undefined;
         },
       };
     },

packages/core/src/plugins/use-payload-formatter.ts

@@ -1,15 +1,33 @@
 import { Plugin } from '@envelop/types';
 import { ExecutionResult } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
-export const usePayloadFormatter = (formatter: (result: ExecutionResult<any, any>) => false | ExecutionResult<any, any>): Plugin => ({
+export type FormatterFunction = (result: ExecutionResult<any, any>) => false | ExecutionResult<any, any>;
+
+const makeHandleResult =
+  (formatter: FormatterFunction) =>
+  ({ result, setResult }: { result: ExecutionResult; setResult: (result: ExecutionResult) => void }) => {
+    const modified = formatter(result);
+    if (modified !== false) {
+      setResult(modified);
+    }
+  };
+
+export const usePayloadFormatter = (formatter: FormatterFunction): Plugin => ({
   onExecute() {
+    const handleResult = makeHandleResult(formatter);
     return {
-      onExecuteDone: ({ result, setResult }) => {
-        const modified = formatter(result);
-
-        if (modified !== false) {
-          setResult(modified);
+      onExecuteDone({ result, setResult }) {
+        if (isAsyncIterable(result)) {
+          return {
+            onNext({ result, setResult }) {
+              handleResult({ result, setResult });
+            },
+          };
         }
+
+        handleResult({ result, setResult });
+        return undefined;
       },
     };
   },