fix: subscriptions & webhooks (#7816)

* fix: subscriptions & webhooks * chore(dependencies): updated changesets for modified dependencies * Fix lockfile * Fix TS * pipeLogs: true remove --------- Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
ardatan · Oct 23, 2024 · fad4d27 · fad4d27
1 parent e5af4f9
commit fad4d27
Show file tree

Hide file tree

Showing 16 changed files with 217 additions and 90 deletions.
diff --git a/.changeset/@graphql-mesh_fusion-runtime-7816-dependencies.md b/.changeset/@graphql-mesh_fusion-runtime-7816-dependencies.md
@@ -0,0 +1,5 @@
+---
+"@graphql-mesh/fusion-runtime": patch
+---
+dependencies updates:
+  - Added dependency [`@graphql-tools/merge@^9.0.8` ↗︎](https://www.npmjs.com/package/@graphql-tools/merge/v/9.0.8) (to `dependencies`)
diff --git a/.changeset/mean-papayas-look.md b/.changeset/mean-papayas-look.md
@@ -0,0 +1,8 @@
+---
+'@graphql-mesh/fusion-composition': patch
+'@graphql-mesh/fusion-runtime': patch
+'@graphql-mesh/config': patch
+'@graphql-mesh/utils': patch
+---
+
+Fix subscription fields fed by webhooks and defined in \`additionalTypeDefs\` with \`@resolveTo\`
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -126,7 +126,7 @@ jobs:
           command: yarn test:leak
 
   integration:
-    needs: [unit]
+    needs: [typecheck, unit]
     name: integration / node ${{matrix.node-version}}
     timeout-minutes: 60
     runs-on: ubuntu-latest

diff --git a/e2e/json-schema-subscriptions/__snapshots__/json-schema-subscriptions.test.ts.snap b/e2e/json-schema-subscriptions/__snapshots__/json-schema-subscriptions.test.ts.snap
@@ -1,7 +1,7 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
 exports[`should compose the appropriate schema 1`] = `
-"schema @link(url: "https://specs.apollo.dev/link/v1.0") @link(url: "https://specs.apollo.dev/join/v0.3", for: EXECUTION) @link(url: "https://the-guild.dev/graphql/mesh/spec/v1.0", import: ["@example", "@httpOperation", "@pubsubOperation", "@transport", "@extraSchemaDefinitionDirective"]) {
+"schema @link(url: "https://the-guild.dev/graphql/mesh/spec/v1.0") {
   query: Query
   mutation: Mutation
   subscription: Subscription
@@ -31,6 +31,8 @@ directive @transport(subgraph: String, kind: String, location: String, headers:
 
 directive @extraSchemaDefinitionDirective(directives: _DirectiveExtensions) repeatable on OBJECT
 
+directive @additionalField on FIELD_DEFINITION
+
 directive @live on QUERY
 
 scalar join__FieldSet
@@ -67,24 +69,19 @@ type query_todos_items @join__type(graph: API) {
 }
 
 type Mutation @join__type(graph: API) {
-  addTodo(input: mutationInput_addTodo_input_Input): mutation_addTodo @httpOperation(subgraph: "API", path: "/todo", httpMethod: POST)
+  addTodo(input: mutationInput_addTodo_input_Input): Todo @httpOperation(subgraph: "API", path: "/todo", httpMethod: POST)
 }
 
-type mutation_addTodo @example(subgraph: "API", value: "{\\"id\\":0,\\"name\\":\\"TodoName\\",\\"content\\":\\"TodoContent\\"}") @join__type(graph: API) {
+type Todo @example(subgraph: "API", value: "{\\"id\\":0,\\"name\\":\\"TodoName\\",\\"content\\":\\"TodoContent\\"}") @join__type(graph: API) {
   id: Int
   name: String
   content: String
 }
 
 type Subscription @join__type(graph: API) {
   """PubSub Topic: webhook:post:/webhooks/todo_added"""
-  todoAdded: subscription_todoAdded @pubsubOperation(subgraph: "API", pubsubTopic: "webhook:post:/webhooks/todo_added")
-}
-
-type subscription_todoAdded @example(subgraph: "API", value: "{\\"id\\":0,\\"name\\":\\"TodoName\\",\\"content\\":\\"TodoContent\\"}") @join__type(graph: API) {
-  id: Int
-  name: String
-  content: String
+  todoAddedFromSource: Todo @pubsubOperation(subgraph: "API", pubsubTopic: "webhook:post:/webhooks/todo_added")
+  todoAddedFromExtensions: Todo @resolveTo(pubsubTopic: "webhook:post:/webhooks/todo_added") @additionalField
 }
 
 enum HTTPMethod @join__type(graph: API) {

diff --git a/e2e/json-schema-subscriptions/json-schema-subscriptions.test.ts b/e2e/json-schema-subscriptions/json-schema-subscriptions.test.ts
@@ -9,60 +9,62 @@ it('should compose the appropriate schema', async () => {
   expect(result).toMatchSnapshot();
 });
 
-it('should query, mutate and subscribe', async () => {
-  const servePort = await getAvailablePort();
-  const api = await service('api', { servePort });
-  const { output } = await compose({ output: 'graphql', services: [api] });
-  const { execute } = await serve({ supergraph: output, port: servePort });
+['todoAddedFromSource', 'todoAddedFromExtensions'].forEach(subscriptionField => {
+  describe(`Listen to ${subscriptionField}`, () => {
+    it('should query, mutate and subscribe', async () => {
+      const servePort = await getAvailablePort();
+      const api = await service('api', { servePort });
+      const { output } = await compose({ output: 'graphql', services: [api] });
+      const { execute } = await serve({ supergraph: output, port: servePort });
 
-  await expect(
-    execute({
-      query: /* GraphQL */ `
-        query Todos {
-          todos {
-            name
-            content
-          }
-        }
-      `,
-    }),
-  ).resolves.toMatchInlineSnapshot(`
+      await expect(
+        execute({
+          query: /* GraphQL */ `
+            query Todos {
+              todos {
+                name
+                content
+              }
+            }
+          `,
+        }),
+      ).resolves.toMatchInlineSnapshot(`
 {
   "data": {
     "todos": [],
   },
 }
 `);
 
-  const sse = createClient({
-    url: `http://localhost:${servePort}/graphql`,
-    retryAttempts: 0,
-    fetchFn: fetch,
-  });
+      const sse = createClient({
+        url: `http://localhost:${servePort}/graphql`,
+        retryAttempts: 0,
+        fetchFn: fetch,
+      });
 
-  const sub = sse.iterate({
-    query: /* GraphQL */ `
-      subscription TodoAdded {
-        todoAdded {
+      const sub = sse.iterate({
+        query: /* GraphQL */ `
+      subscription ${subscriptionField} {
+        ${subscriptionField} {
           name
           content
         }
       }
     `,
-  });
+      });
 
-  await expect(
-    execute({
-      query: /* GraphQL */ `
-        mutation AddTodo {
-          addTodo(input: { name: "Shopping", content: "Buy Milk" }) {
-            name
-            content
-          }
-        }
-      `,
-    }),
-  ).resolves.toMatchInlineSnapshot(`
+      await expect(
+        execute({
+          query: /* GraphQL */ `
+            mutation AddTodo {
+              addTodo(input: { name: "Shopping", content: "Buy Milk" }) {
+                name
+                content
+              }
+            }
+          `,
+        }),
+      ).resolves.toMatchInlineSnapshot(`
 {
   "data": {
     "addTodo": {
@@ -73,17 +75,19 @@ it('should query, mutate and subscribe', async () => {
 }
 `);
 
-  for await (const msg of sub) {
-    expect(msg).toMatchInlineSnapshot(`
+      for await (const msg of sub) {
+        expect(msg).toMatchInlineSnapshot(`
 {
   "data": {
-    "todoAdded": {
+    "${subscriptionField}": {
       "content": "Buy Milk",
       "name": "Shopping",
     },
   },
 }
 `);
-    break;
-  }
+        break;
+      }
+    });
+  });
 });
diff --git a/e2e/json-schema-subscriptions/mesh.config.ts b/e2e/json-schema-subscriptions/mesh.config.ts
@@ -30,19 +30,31 @@ export const composeConfig = defineComposeConfig({
             method: 'POST',
             requestSample: './addTodo.json',
             responseSample: './todo.json',
+            responseTypeName: 'Todo',
           },
           {
             type: OperationTypeNode.SUBSCRIPTION,
-            field: 'todoAdded',
+            field: 'todoAddedFromSource',
             pubsubTopic: 'webhook:post:/webhooks/todo_added',
             responseSample: './todo.json',
+            responseTypeName: 'Todo',
           },
         ],
       }),
     },
   ],
   additionalTypeDefs: /* GraphQL */ `
     directive @live on QUERY
+
+    # If you don't have Subscription type defined anywhere
+    # You have to extend subscription definition
+    extend schema {
+      subscription: Subscription
+    }
+
+    type Subscription {
+      todoAddedFromExtensions: Todo @resolveTo(pubsubTopic: "webhook:post:/webhooks/todo_added")
+    }
   `,
 });
 

diff --git a/packages/fusion/composition/src/transforms/encapsulate.ts b/packages/fusion/composition/src/transforms/encapsulate.ts
@@ -132,9 +132,9 @@ export const resolveToDirective = new GraphQLDirective({
     result: { type: GraphQLString },
     resultType: { type: GraphQLString },
     sourceArgs: { type: resolveToSourceArgsScalar },
-    sourceFieldName: { type: new GraphQLNonNull(GraphQLString) },
-    sourceName: { type: new GraphQLNonNull(GraphQLString) },
+    sourceFieldName: { type: GraphQLString },
+    sourceName: { type: GraphQLString },
     sourceSelectionSet: { type: GraphQLString },
-    sourceTypeName: { type: new GraphQLNonNull(GraphQLString) },
+    sourceTypeName: { type: GraphQLString },
   },
 });
diff --git a/packages/fusion/runtime/package.json b/packages/fusion/runtime/package.json
@@ -59,6 +59,7 @@
     "@graphql-tools/delegate": "^10.0.26",
     "@graphql-tools/executor": "^1.3.2",
     "@graphql-tools/federation": "^2.2.17",
+    "@graphql-tools/merge": "^9.0.8",
     "@graphql-tools/stitch": "^9.2.15",
     "@graphql-tools/stitching-directives": "^3.1.7",
     "@graphql-tools/utils": "^10.5.5",

diff --git a/packages/fusion/runtime/src/federation/subgraph.ts b/packages/fusion/runtime/src/federation/subgraph.ts
@@ -40,7 +40,6 @@ export interface HandleFederationSubschemaOpts {
   schemaDirectives?: Record<string, any>;
   transportEntryMap: Record<string, TransportEntry>;
   additionalTypeDefs: TypeSource[];
-  additionalResolvers: IResolvers<unknown, any>[];
   stitchingDirectivesTransformer: (subschemaConfig: SubschemaConfig) => SubschemaConfig;
   onSubgraphExecute: ReturnType<typeof getOnSubgraphExecute>;
 }
@@ -51,7 +50,6 @@ export function handleFederationSubschema({
   schemaDirectives,
   transportEntryMap,
   additionalTypeDefs,
-  additionalResolvers,
   stitchingDirectivesTransformer,
   onSubgraphExecute,
 }: HandleFederationSubschemaOpts) {
@@ -140,15 +138,6 @@ export function handleFederationSubschema({
             },
           ],
         });
-        for (const resolveToDirective of resolveToDirectives) {
-          additionalResolvers.push(
-            resolveAdditionalResolversWithoutImport({
-              targetTypeName: typeName,
-              targetFieldName: fieldName,
-              ...resolveToDirective,
-            }),
-          );
-        }
       }
       const additionalFieldDirectives = fieldDirectives.additionalField;
       if (additionalFieldDirectives?.length > 0) {

diff --git a/packages/fusion/runtime/src/federation/supergraph.ts b/packages/fusion/runtime/src/federation/supergraph.ts
@@ -1,8 +1,17 @@
-import { isEnumType, Kind, visit, type GraphQLSchema } from 'graphql';
+import {
+  isEnumType,
+  Kind,
+  print,
+  visit,
+  type GraphQLSchema,
+  type ObjectTypeDefinitionNode,
+} from 'graphql';
 import type { TransportEntry } from '@graphql-mesh/transport-common';
+import type { YamlConfig } from '@graphql-mesh/types';
 import { resolveAdditionalResolversWithoutImport } from '@graphql-mesh/utils';
 import type { SubschemaConfig } from '@graphql-tools/delegate';
 import { getStitchedSchemaFromSupergraphSdl } from '@graphql-tools/federation';
+import { mergeTypeDefs } from '@graphql-tools/merge';
 import { stitchingDirectives } from '@graphql-tools/stitching-directives';
 import {
   asArray,
@@ -122,25 +131,46 @@ export const handleFederationSupergraph: UnifiedGraphHandler = function ({
         schemaDirectives,
         transportEntryMap,
         additionalTypeDefs,
-        additionalResolvers,
         stitchingDirectivesTransformer,
         onSubgraphExecute,
       }),
     batch,
     onStitchingOptions(opts: any) {
       subschemas = opts.subschemas;
-      opts.typeDefs = [opts.typeDefs, additionalTypeDefs];
+      const mergedTypeDefs = mergeTypeDefs([opts.typeDefs, additionalTypeDefs]);
+      visit(mergedTypeDefs, {
+        [Kind.FIELD_DEFINITION](field, _key, _parent, _path, ancestors) {
+          const fieldDirectives = getDirectiveExtensions<{
+            resolveTo: YamlConfig.AdditionalStitchingResolverObject;
+          }>({ astNode: field });
+          const resolveToDirectives = fieldDirectives?.resolveTo;
+          if (resolveToDirectives?.length) {
+            const targetTypeName = (ancestors[ancestors.length - 1] as ObjectTypeDefinitionNode)
+              .name.value;
+            const targetFieldName = field.name.value;
+            for (const resolveToDirective of resolveToDirectives) {
+              additionalResolvers.push(
+                resolveAdditionalResolversWithoutImport({
+                  targetTypeName,
+                  targetFieldName,
+                  ...resolveToDirective,
+                }),
+              );
+            }
+          }
+        },
+      });
+      opts.typeDefs = mergedTypeDefs;
       opts.resolvers = additionalResolvers;
     },
-    onSubgraphAST(name, subgraphAST) {
+    onSubgraphAST(_name, subgraphAST) {
       return visit(subgraphAST, {
         [Kind.OBJECT_TYPE_DEFINITION](node) {
           const typeName = node.name.value;
           return {
             ...node,
             fields: node.fields.filter(fieldNode => {
               const fieldDirectives = getDirectiveExtensions({ astNode: fieldNode });
-              const fieldName = fieldNode.name.value;
               const resolveToDirectives = fieldDirectives.resolveTo;
               if (resolveToDirectives?.length > 0) {
                 additionalTypeDefs.push({
@@ -153,15 +183,6 @@ export const handleFederationSupergraph: UnifiedGraphHandler = function ({
                     },
                   ],
                 });
-                for (const resolveToDirective of resolveToDirectives) {
-                  additionalResolvers.push(
-                    resolveAdditionalResolversWithoutImport({
-                      targetTypeName: typeName,
-                      targetFieldName: fieldName,
-                      ...(resolveToDirective as any),
-                    }),
-                  );
-                }
               }
               const additionalFieldDirectives = fieldDirectives.additionalField;
               if (additionalFieldDirectives?.length > 0) {

diff --git a/packages/legacy/config/src/process.ts b/packages/legacy/config/src/process.ts
@@ -468,9 +468,9 @@ export async function processConfig(
         scalar ResolveToSourceArgs
         directive @resolveTo(
           requiredSelectionSet: String
-          sourceName: String!
-          sourceTypeName: String!
-          sourceFieldName: String!
+          sourceName: String
+          sourceTypeName: String
+          sourceFieldName: String
           sourceSelectionSet: String
           sourceArgs: ResolveToSourceArgs
           keyField: String