fix: forward operation directives to the subschema requests (#6420)

* fix: forward operation directives to the subschema requests

* chore(dependencies): updated changesets for modified dependencies

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: ardatan

.changeset/@graphql-tools_delegate-6420-dependencies.md

@@ -0,0 +1,5 @@
+---
+"@graphql-tools/delegate": patch
+---
+dependencies updates:
+  - Added dependency [`@repeaterjs/repeater@^3.0.6` ↗︎](https://www.npmjs.com/package/@repeaterjs/repeater/v/3.0.6) (to `dependencies`)

.changeset/good-parents-push.md

@@ -0,0 +1,10 @@
+---
+'@graphql-tools/delegate': patch
+---
+
+Pass operation directives correctly to the subschema;
+```graphql
+query {
+  hello @someDir
+}
+```

.changeset/nice-parrots-kick.md

@@ -0,0 +1,6 @@
+---
+'@graphql-tools/executor': patch
+'@graphql-tools/utils': patch
+---
+
+`mapAsyncIterator` now accepts `AsyncIterable`

packages/delegate/package.json

@@ -54,6 +54,7 @@
     "@graphql-tools/executor": "^1.3.0",
     "@graphql-tools/schema": "^10.0.4",
     "@graphql-tools/utils": "^10.2.3",
+    "@repeaterjs/repeater": "^3.0.6",
     "dataloader": "^2.2.2",
     "tslib": "^2.5.0"
   },

packages/delegate/src/createRequest.ts

@@ -107,8 +107,8 @@ export function createRequest({
       newVariables,
     );
   }
-
-  const rootFieldName = targetFieldName ?? fieldNodes?.[0]?.name.value;
+  const fieldNode = fieldNodes?.[0];
+  const rootFieldName = targetFieldName ?? fieldNode?.name.value;
 
   if (rootFieldName === undefined) {
     throw new Error(`Either "targetFieldName" or a non empty "fieldNodes" array must be provided.`);
@@ -122,6 +122,7 @@ export function createRequest({
       value: rootFieldName,
     },
     selectionSet: newSelectionSet,
+    directives: fieldNode?.directives,
   };
 
   const operationName: NameNode | undefined = targetOperationName

packages/delegate/src/delegateToSchema.ts

@@ -3,6 +3,7 @@ import {
   FieldDefinitionNode,
   GraphQLOutputType,
   GraphQLSchema,
+  isListType,
   OperationTypeNode,
   validate,
 } from 'graphql';
@@ -21,6 +22,7 @@ import {
   MaybeAsyncIterable,
   memoize1,
 } from '@graphql-tools/utils';
+import { Repeater } from '@repeaterjs/repeater';
 import { applySchemaTransforms } from './applySchemaTransforms.js';
 import { createRequest, getDelegatingOperation } from './createRequest.js';
 import { Subschema } from './Subschema.js';
@@ -105,9 +107,44 @@ export function delegateRequest<
 
   function handleExecutorResult(executorResult: MaybeAsyncIterable<ExecutionResult<any>>) {
     if (isAsyncIterable(executorResult)) {
-      const iterator = executorResult[Symbol.asyncIterator]();
-      // "subscribe" to the subscription result and map the result through the transforms
-      return mapAsyncIterator(iterator, result => transformer.transformResult(result));
+      // This might be a stream
+      if (delegationContext.operation === 'query' && isListType(delegationContext.returnType)) {
+        return new Repeater<ExecutionResult<any>>(async (push, stop) => {
+          const pushed = new WeakSet();
+          let stopped = false;
+          stop.finally(() => {
+            stopped = true;
+          });
+          try {
+            for await (const result of executorResult) {
+              if (stopped) {
+                break;
+              }
+              const transformedResult = await transformer.transformResult(result);
+              // @stream needs to get the results one by one
+              if (Array.isArray(transformedResult)) {
+                for (const individualResult$ of transformedResult) {
+                  if (stopped) {
+                    break;
+                  }
+                  const individualResult = await individualResult$;
+                  // Avoid pushing the same result multiple times
+                  if (!pushed.has(individualResult)) {
+                    pushed.add(individualResult);
+                    await push(individualResult);
+                  }
+                }
+              } else {
+                await push(await transformedResult);
+              }
+            }
+            stop();
+          } catch (error) {
+            stop(error);
+          }
+        });
+      }
+      return mapAsyncIterator(executorResult, result => transformer.transformResult(result));
     }
     return transformer.transformResult(executorResult);
   }

packages/executor/src/execution/execute.ts

@@ -1641,7 +1641,7 @@ function mapSourceToResponse(
   // "ExecuteQuery" algorithm, for which `execute` is also used.
   return flattenAsyncIterable(
     mapAsyncIterator(
-      resultOrStream[Symbol.asyncIterator](),
+      resultOrStream,
       async (payload: unknown) =>
         ensureAsyncIterable(await executeImpl(buildPerEventExecutionContext(exeContext, payload))),
       (error: Error) => {

packages/federation/test/__snapshots__/defer-stream.test.ts.snap

@@ -291,7 +291,7 @@ exports[`Defer/Stream streams: stream 1`] = `
     ],
   },
   {
-    "hasNext": false,
+    "hasNext": true,
     "incremental": [
       {
         "items": [
@@ -317,5 +317,8 @@ exports[`Defer/Stream streams: stream 1`] = `
       },
     ],
   },
+  {
+    "hasNext": false,
+  },
 ]
 `;

packages/federation/test/defer-stream.test.ts

@@ -1,11 +1,13 @@
 import { inspect } from 'util';
 import { GraphQLSchema, parse, print } from 'graphql';
+import { createYoga } from 'graphql-yoga';
 import _ from 'lodash';
 import { IntrospectAndCompose, LocalGraphQLDataSource } from '@apollo/gateway';
 import { buildSubgraphSchema } from '@apollo/subgraph';
-import { createDefaultExecutor } from '@graphql-tools/delegate';
 import { normalizedExecutor } from '@graphql-tools/executor';
+import { buildHTTPExecutor } from '@graphql-tools/executor-http';
 import { asArray, ExecutionResult, mergeDeep } from '@graphql-tools/utils';
+import { useDeferStream } from '@graphql-yoga/plugin-defer-stream';
 import { assertAsyncIterable } from '../../loaders/url/tests/test-utils';
 import { getStitchedSchemaFromSupergraphSdl } from '../src/supergraph';
 
@@ -98,6 +100,10 @@ describe('Defer/Stream', () => {
       },
     },
   });
+  const usersServer = createYoga({
+    schema: usersSubgraph,
+    plugins: [useDeferStream()],
+  });
   const postsSubgraph = buildSubgraphSchema({
     typeDefs: parse(/* GraphQL */ `
       type Query {
@@ -141,6 +147,10 @@ describe('Defer/Stream', () => {
       },
     },
   });
+  const postsServer = createYoga({
+    schema: postsSubgraph,
+    plugins: [useDeferStream()],
+  });
   let schema: GraphQLSchema;
   let finalResult: ExecutionResult;
   beforeAll(async () => {
@@ -163,7 +173,10 @@ describe('Defer/Stream', () => {
       onSubschemaConfig(subschemaConfig) {
         const subgraphName = subschemaConfig.name.toLowerCase();
         if (subgraphName === 'users') {
-          const origExecutor = createDefaultExecutor(usersSubgraph);
+          const origExecutor = buildHTTPExecutor({
+            endpoint: 'http://localhost:4001/graphql',
+            fetch: usersServer.fetch,
+          });
           subschemaConfig.executor = async function usersExecutor(execReq) {
             const result = await origExecutor(execReq);
             if (process.env['DEBUG']) {
@@ -176,7 +189,10 @@ describe('Defer/Stream', () => {
             return result;
           };
         } else if (subgraphName === 'posts') {
-          const origExecutor = createDefaultExecutor(postsSubgraph);
+          const origExecutor = buildHTTPExecutor({
+            endpoint: 'http://localhost:4002/graphql',
+            fetch: postsServer.fetch,
+          });
           subschemaConfig.executor = async function postsExecutor(execReq) {
             const result = await origExecutor(execReq);
             if (process.env['DEBUG']) {

packages/stitch/tests/operationDirectives.test.ts

@@ -0,0 +1,48 @@
+import { parse, print } from 'graphql';
+import { normalizedExecutor } from '@graphql-tools/executor';
+import { makeExecutableSchema } from '@graphql-tools/schema';
+import { stitchSchemas } from '../src/stitchSchemas';
+
+describe('Operation Directives', () => {
+  it('sends the directives to the subschema operations', async () => {
+    const schema = makeExecutableSchema({
+      typeDefs: /* GraphQL */ `
+        directive @strExpr on FIELD
+        type Query {
+          hello: String
+        }
+      `,
+      resolvers: {
+        Query: {
+          hello: (_source, _args, _context, info) => {
+            return print(info.operation);
+          },
+        },
+      },
+    });
+
+    const gateway = stitchSchemas({
+      subschemas: [schema],
+      mergeDirectives: true,
+    });
+
+    const res = await normalizedExecutor({
+      schema: gateway,
+      document: parse(/* GraphQL */ `
+        query getHello {
+          hello @strExpr
+        }
+      `),
+    });
+
+    expect(res).toEqual({
+      data: {
+        hello: /* GraphQL */ `
+query getHello {
+  __typename
+  hello @strExpr
+}`.trim(),
+      },
+    });
+  });
+});