refactor batchDelegateToSchema

...to use DataLoader for the actual GraphQL results, rather than the external values

This allows us to properly transduce streams.

TODO:
1. We should now be able to add the stream directive for batchDelegateToSchema.
2. Refactor error parsing, now we are relying on the onLocatedError option, but this
   is no longer necessary. Note that additional tests for batched errors are now passing
   (see #2951)
3. Bring back createBatchDelegateFn, that may be useful
4. The Receiver is now created separately for each list item, which means that the
   initialPathDepth probably does not match? This will require a separate argument to
   the Receiver constructor for adjustment
5. Bring back valuesFromResults and fix documentation, we now operate on GraphQL
   results, so should be fewer pitfalls (see #2829)

Author: yaacovCR

packages/batch-delegate/src/batchDelegateToSchema.ts

@@ -1,14 +1,89 @@
 import { BatchDelegateOptions } from './types';
 
+import { AsyncExecutionResult, ExecutionResult, getNullableType, GraphQLList } from 'graphql';
+
+import {
+  DelegationContext,
+  createRequestFromInfo,
+  externalValueFromResult,
+  getDelegationContext,
+  getDelegatingOperation,
+  Receiver,
+} from '@graphql-tools/delegate';
+
+import { isAsyncIterable, relocatedError } from '@graphql-tools/utils';
+
 import { getLoader } from './getLoader';
 
-export function batchDelegateToSchema(options: BatchDelegateOptions): any {
+export async function batchDelegateToSchema(options: BatchDelegateOptions): Promise<any> {
   const key = options.key;
   if (key == null) {
     return null;
   } else if (Array.isArray(key) && !key.length) {
     return [];
   }
-  const loader = getLoader(options);
-  return Array.isArray(key) ? loader.loadMany(key) : loader.load(key);
+
+  const {
+    info,
+    operationName,
+    operation = getDelegatingOperation(info.parentType, info.schema),
+    fieldName = info.fieldName,
+    returnType = info.returnType,
+    selectionSet,
+    fieldNodes,
+  } = options;
+
+  if (operation !== 'query' && operation !== 'mutation') {
+    throw new Error(`Batch delegation not possible for operation '${operation}'.`);
+  }
+
+  const request = createRequestFromInfo({
+    info,
+    operation,
+    fieldName,
+    selectionSet,
+    fieldNodes,
+    operationName,
+  });
+
+  const delegationContext = getDelegationContext({
+    request,
+    onLocatedError: originalError => relocatedError(originalError, originalError.path.slice(1)),
+    ...options,
+    operation,
+    fieldName,
+    returnType,
+  });
+
+  const loader = getLoader(options, request, delegationContext);
+
+  if (Array.isArray(key)) {
+    const results = await loader.loadMany(key);
+
+    return results.map(result =>
+      onResult(result, {
+        ...delegationContext,
+        returnType: (getNullableType(delegationContext.returnType) as GraphQLList<any>).ofType,
+      })
+    );
+  }
+
+  const result = await loader.load(key);
+  return onResult(result, delegationContext);
+}
+
+function onResult(
+  result: Error | ExecutionResult | AsyncIterableIterator<AsyncExecutionResult>,
+  delegationContext: DelegationContext
+): any {
+  if (result instanceof Error) {
+    return result;
+  }
+
+  if (isAsyncIterable(result)) {
+    const receiver = new Receiver(result, delegationContext);
+    return receiver.getInitialValue();
+  }
+
+  return externalValueFromResult(result, delegationContext);
 }

packages/batch-delegate/src/createBatchDelegateFn.ts

@@ -1,19 +1,17 @@
-import { getNamedType, GraphQLOutputType, GraphQLList, GraphQLSchema, FieldNode } from 'graphql';
+import { GraphQLSchema, FieldNode } from 'graphql';
 
 import DataLoader from 'dataloader';
 
+import { SubschemaConfig, Transformer, getExecutor, validateRequest, DelegationContext } from '@graphql-tools/delegate';
 import {
-  SubschemaConfig,
-  Transformer,
-  createRequestFromInfo,
-  getDelegationContext,
-  getDelegatingOperation,
-  getExecutor,
-  validateRequest,
-  Receiver,
-  externalValueFromResult,
-} from '@graphql-tools/delegate';
-import { isAsyncIterable, relocatedError } from '@graphql-tools/utils';
+  AsyncExecutionResult,
+  ExecutionPatchResult,
+  ExecutionResult,
+  isAsyncIterable,
+  mapAsyncIterator,
+  Request,
+  splitAsyncIterator,
+} from '@graphql-tools/utils';
 
 import { BatchDelegateOptions } from './types';
 
@@ -22,48 +20,21 @@ const cache1: WeakMap<
   WeakMap<GraphQLSchema | SubschemaConfig, Record<string, DataLoader<any, any>>>
 > = new WeakMap();
 
-function createBatchFn<K = any>(options: BatchDelegateOptions) {
+function createBatchFn<K = any>(options: BatchDelegateOptions, request: Request, delegationContext: DelegationContext) {
   const argsFromKeys = options.argsFromKeys ?? ((keys: ReadonlyArray<K>) => ({ ids: keys }));
-  const { lazyOptionsFn } = options;
 
-  return async (keys: ReadonlyArray<K>) => {
-    const {
-      context,
-      info,
-      operationName,
-      operation = getDelegatingOperation(info.parentType, info.schema),
-      fieldName = info.fieldName,
-      returnType = new GraphQLList(getNamedType(options.info.returnType) as GraphQLOutputType),
-      selectionSet,
-      fieldNodes,
-      binding,
-      skipValidation,
-    } = options;
-
-    if (operation !== 'query' && operation !== 'mutation') {
-      throw new Error(`Batch delegation not possible for operation '${operation}'.`);
-    }
+  const { binding, skipValidation } = options;
 
-    const request = createRequestFromInfo({
-      info,
-      operation,
-      fieldName,
-      selectionSet,
-      fieldNodes,
-      operationName,
-    });
+  const { fieldName, context, info } = delegationContext;
 
-    const delegationContext = getDelegationContext({
-      request,
-      args: argsFromKeys(keys),
-      onLocatedError: originalError => relocatedError(originalError, originalError.path.slice(1)),
-      ...(lazyOptionsFn == null ? options : lazyOptionsFn(options)),
-      operation,
-      fieldName,
-      returnType,
-    });
-
-    const transformer = new Transformer(delegationContext, binding);
+  return async (keys: ReadonlyArray<K>) => {
+    const transformer = new Transformer(
+      {
+        ...delegationContext,
+        args: argsFromKeys(keys),
+      },
+      binding
+    );
 
     const processedRequest = transformer.transformRequest(request);
 
@@ -79,40 +50,38 @@ function createBatchFn<K = any>(options: BatchDelegateOptions) {
       info,
     });
 
+    const numKeys = keys.length;
     if (isAsyncIterable(batchResult)) {
-      // TODO: split the asyncIterable and make a new receiver from each of them, return the Receiver instead of the
-      // initial value, so that the correct info can be used to instantiate the Receiver
-      const receiver = new Receiver(batchResult, delegationContext, executionResult =>
-        transformer.transformResult(executionResult)
-      );
-
-      const batchValue = await receiver.getInitialValue();
-
-      return Array.isArray(batchValue) ? batchValue : keys.map(() => batchValue);
+      const mappedBatchResult = mapAsyncIterator(batchResult, result => transformer.transformResult(result));
+      return splitAsyncIterator(mappedBatchResult, numKeys, result => splitAsyncResult(result, fieldName));
     }
 
-    // TODO: split the batchedResult and return the result instead of the value, so the correct info
-    // can be used to instantiate the value
-    const batchValue = externalValueFromResult(transformer.transformResult(batchResult), delegationContext);
-
-    return Array.isArray(batchValue) ? batchValue : keys.map(() => batchValue);
+    return splitResult(transformer.transformResult(batchResult), fieldName, numKeys);
   };
 }
 
-export function getLoader<K = any, V = any, C = K>(options: BatchDelegateOptions): DataLoader<K, V, C> {
+export function getLoader<K = any, C = K>(
+  options: BatchDelegateOptions,
+  request: Request,
+  delegationContext: DelegationContext
+): DataLoader<K, ExecutionResult | AsyncIterableIterator<AsyncExecutionResult>, C> {
   const fieldName = options.fieldName ?? options.info.fieldName;
 
-  let cache2: WeakMap<GraphQLSchema | SubschemaConfig, Record<string, DataLoader<K, V, C>>> = cache1.get(
-    options.info.fieldNodes
-  );
+  let cache2: WeakMap<
+    GraphQLSchema | SubschemaConfig,
+    Record<string, DataLoader<K, ExecutionResult | AsyncIterableIterator<AsyncExecutionResult>, C>>
+  > = cache1.get(options.info.fieldNodes);
 
   if (cache2 === undefined) {
     cache2 = new WeakMap();
     cache1.set(options.info.fieldNodes, cache2);
     const loaders = Object.create(null);
     cache2.set(options.schema, loaders);
-    const batchFn = createBatchFn(options);
-    const loader = new DataLoader<K, V, C>(keys => batchFn(keys), options.dataLoaderOptions);
+    const batchFn = createBatchFn(options, request, delegationContext);
+    const loader = new DataLoader<K, ExecutionResult | AsyncIterableIterator<AsyncExecutionResult>, C>(
+      keys => batchFn(keys),
+      options.dataLoaderOptions
+    );
     loaders[fieldName] = loader;
     return loader;
   }
@@ -122,19 +91,96 @@ export function getLoader<K = any, V = any, C = K>(options: BatchDelegateOptions
   if (loaders === undefined) {
     loaders = Object.create(null);
     cache2.set(options.schema, loaders);
-    const batchFn = createBatchFn(options);
-    const loader = new DataLoader<K, V, C>(keys => batchFn(keys), options.dataLoaderOptions);
+    const batchFn = createBatchFn(options, request, delegationContext);
+    const loader = new DataLoader<K, ExecutionResult | AsyncIterableIterator<AsyncExecutionResult>, C>(
+      keys => batchFn(keys),
+      options.dataLoaderOptions
+    );
     loaders[fieldName] = loader;
     return loader;
   }
 
   let loader = loaders[fieldName];
 
   if (loader === undefined) {
-    const batchFn = createBatchFn(options);
-    loader = new DataLoader<K, V, C>(keys => batchFn(keys), options.dataLoaderOptions);
+    const batchFn = createBatchFn(options, request, delegationContext);
+    loader = new DataLoader<K, ExecutionResult | AsyncIterableIterator<AsyncExecutionResult>, C>(
+      keys => batchFn(keys),
+      options.dataLoaderOptions
+    );
     loaders[fieldName] = loader;
   }
 
   return loader;
 }
+
+function splitResult(result: ExecutionResult, fieldName: string, numItems: number): Array<ExecutionResult> {
+  const { data, errors } = result;
+  const fieldData = data?.[fieldName];
+
+  if (fieldData === undefined) {
+    if (errors === undefined) {
+      return Array(numItems).fill({});
+    }
+
+    return Array(numItems).fill({ errors });
+  }
+
+  return fieldData.map((value: any) => ({
+    data: {
+      [fieldName]: value,
+    },
+    errors,
+  }));
+}
+
+function splitAsyncResult(result: AsyncExecutionResult, fieldName: string): [[number, AsyncExecutionResult]] {
+  const { data, errors, path } = result as ExecutionPatchResult;
+
+  if (path === undefined || path.length === 0) {
+    const fieldData = data?.[fieldName];
+    if (fieldData !== undefined) {
+      return fieldData.map((value: any, index: number) => [
+        index,
+        {
+          data: {
+            [fieldName]: value,
+          },
+          errors,
+        },
+      ]);
+    }
+  } else if (path[0] === fieldName) {
+    const index = path[1] as number;
+
+    if (path.length === 2) {
+      return [
+        [
+          index,
+          {
+            ...result,
+            data: {
+              [fieldName]: data,
+            },
+            errors,
+          },
+        ],
+      ];
+    }
+
+    const newPath = [fieldName, ...path.slice(2)];
+    return [
+      [
+        index,
+        {
+          ...result,
+          data,
+          errors,
+          path: newPath,
+        },
+      ],
+    ];
+  }
+
+  return [[undefined, result]];
+}

packages/batch-delegate/src/getLoader.ts

@@ -1,4 +1,3 @@
 export * from './batchDelegateToSchema';
-export * from './createBatchDelegateFn';
 
 export * from './types';

packages/batch-delegate/src/index.ts

@@ -15,12 +15,4 @@ export interface BatchDelegateOptions<TContext = Record<string, any>, K = any, V
   dataLoaderOptions?: DataLoader.Options<K, V, C>;
   key: K;
   argsFromKeys?: (keys: ReadonlyArray<K>) => Record<string, any>;
-  lazyOptionsFn?: BatchDelegateOptionsFn;
-}
-
-export interface CreateBatchDelegateFnOptions<TContext = Record<string, any>, K = any, V = any, C = K>
-  extends Partial<Omit<IDelegateToSchemaOptions<TContext>, 'args' | 'info'>> {
-  dataLoaderOptions?: DataLoader.Options<K, V, C>;
-  argsFromKeys?: (keys: ReadonlyArray<K>) => Record<string, any>;
-  lazyOptionsFn?: (batchDelegateOptions: BatchDelegateOptions<TContext, K>) => IDelegateToSchemaOptions<TContext>;
 }