Support Type Merging within additional type defs with subscriptions (#8733)

Author: ardatan

.changeset/solid-needles-shout.md

@@ -0,0 +1,79 @@
+---
+'@graphql-mesh/config': patch
+'@graphql-mesh/types': patch
+'@graphql-mesh/utils': patch
+---
+
+Support Type Merging within additional type defs for subscriptions
+
+This allows subscription events to resolve fields from other subgraphs.
+
+For example, if you have a `products` subgraph like this:
+
+```gql filename="products.graphql"
+type Query {
+  hello: String!
+}
+type Product @key(fields: "id") {
+  id: ID!
+  name: String!
+  price: Float!
+}
+```
+
+we need add the subscription fields like this:
+
+```ts filename="mesh.config.ts"
+import { defineConfig, loadGraphQLHTTPSubgraph } from '@graphql-mesh/compose-cli'
+
+export const composeConfig = defineConfig({
+  subgraphs: [
+    {
+      sourceHandler: loadGraphQLHTTPSubgraph('products', {
+        endpoint: `http://localhost:3000/graphql`
+      })
+    }
+  ],
+  additionalTypeDefs: /* GraphQL */ `
+    extend schema {
+      subscription: Subscription
+    }
+    type Subscription {
+      newProduct: Product! @resolveTo(pubsubTopic: "new_product", sourceName: "products")
+    }
+  `
+})
+```
+
+you can subscribe to Hive Gateway like this:
+
+```graphql
+subscription {
+  newProduct {
+    name
+    price
+  }
+}
+```
+
+emit an event to the Redis instance on the `new_product` topic
+this:
+
+```redis
+PUBLISH new_product '{"id":"roomba70x"}'
+```
+
+The subscriber will then receive the following event:
+
+```json
+{
+  "data": {
+    "newProduct": {
+      "name": "Roomba 70x",
+      "price": 279.99
+    }
+  }
+}
+```
+
+Because Hive Gateway merged the `Product` type from the `products` subgraph into the root schema, it can resolve the `name` and `price` fields even though they are not defined in the `additionalTypeDefs`.

e2e/subscriptions-type-merging/gateway.config.ts

@@ -0,0 +1,31 @@
+import { Redis } from 'ioredis';
+import { defineConfig as defineGatewayConfig } from '@graphql-hive/gateway';
+import { RedisPubSub } from '@graphql-hive/pubsub/redis';
+import { toMeshPubSub } from '@graphql-mesh/types';
+
+/**
+ * When a Redis connection enters "subscriber mode" (after calling SUBSCRIBE), it can only execute
+ * subscriber commands (SUBSCRIBE, UNSUBSCRIBE, etc.). Meaning, it cannot execute other commands like PUBLISH.
+ * To avoid this, we use two separate Redis clients: one for publishing and one for subscribing.
+ */
+const pub = new Redis({
+  host: process.env['REDIS_HOST'],
+  port: parseInt(process.env['REDIS_PORT']!),
+  autoResubscribe: true,
+});
+const sub = new Redis({
+  host: process.env['REDIS_HOST'],
+  port: parseInt(process.env['REDIS_PORT']!),
+  autoResubscribe: true,
+});
+
+export const gatewayConfig = defineGatewayConfig({
+  pubsub: toMeshPubSub(
+    new RedisPubSub(
+      { pub, sub },
+      {
+        channelPrefix: 'gw',
+      },
+    ),
+  ),
+});

e2e/subscriptions-type-merging/mesh.config.ts

@@ -0,0 +1,22 @@
+import { Opts } from '@e2e/opts';
+import { defineConfig, loadGraphQLHTTPSubgraph } from '@graphql-mesh/compose-cli';
+
+const opts = Opts(process.argv);
+
+export const composeConfig = defineConfig({
+  subgraphs: [
+    {
+      sourceHandler: loadGraphQLHTTPSubgraph('products', {
+        endpoint: `http://localhost:${opts.getServicePort('products')}/graphql`,
+      }),
+    },
+  ],
+  additionalTypeDefs: /* GraphQL */ `
+    extend schema {
+      subscription: Subscription
+    }
+    type Subscription {
+      newProduct: Product! @resolveTo(pubsubTopic: "new_product", sourceName: "products")
+    }
+  `,
+});

e2e/subscriptions-type-merging/package.json

@@ -0,0 +1,11 @@
+{
+  "name": "@e2e/subscriptions-type-merging",
+  "private": true,
+  "dependencies": {
+    "@graphql-hive/gateway": "^1.16.3",
+    "@graphql-hive/pubsub": "next",
+    "graphql": "^16.9.0",
+    "graphql-sse": "^2.5.4",
+    "ioredis": "^5.7.0"
+  }
+}

e2e/subscriptions-type-merging/services/products.ts

@@ -0,0 +1,38 @@
+import { createServer } from 'node:http';
+import { parse } from 'graphql';
+import { createYoga } from 'graphql-yoga';
+import { buildSubgraphSchema } from '@apollo/subgraph';
+import { Opts } from '@e2e/opts';
+
+const port = Opts(process.argv).getServicePort('products');
+
+createServer(
+  createYoga({
+    schema: buildSubgraphSchema({
+      typeDefs: parse(/* GraphQL */ `
+        type Query {
+          hello: String!
+        }
+        type Product @key(fields: "id") {
+          id: ID!
+          name: String!
+          price: Float!
+        }
+      `),
+      resolvers: {
+        Query: {
+          hello: () => 'world',
+        },
+        Product: {
+          __resolveReference: ref => ({
+            id: ref.id,
+            name: `Roomba X${ref.id}`,
+            price: 100,
+          }),
+        },
+      },
+    }),
+  }),
+).listen(port, () => {
+  console.log(`Products subgraph running on http://localhost:${port}/graphql`);
+});

e2e/subscriptions-type-merging/subscriptions-type-merging.test.ts

@@ -0,0 +1,86 @@
+import { setTimeout } from 'node:timers/promises';
+import { createClient } from 'graphql-sse';
+import Redis from 'ioredis';
+import { createTenv } from '@e2e/tenv';
+import { createDeferred } from '@graphql-tools/utils';
+import { AsyncDisposableStack } from '@whatwg-node/disposablestack';
+import { fetch } from '@whatwg-node/fetch';
+
+const { container, compose, service, serve } = createTenv(__dirname);
+
+const redisEnv = {
+  REDIS_HOST: '',
+  REDIS_PORT: 0,
+};
+beforeAll(async () => {
+  const redis = await container({
+    name: 'redis',
+    image: 'redis:8',
+    containerPort: 6379,
+    healthcheck: ['CMD-SHELL', 'redis-cli ping'],
+    env: {
+      LANG: '', // fixes "Failed to configure LOCALE for invalid locale name."
+    },
+  });
+  redisEnv.REDIS_HOST = '0.0.0.0';
+  redisEnv.REDIS_PORT = redis.port;
+});
+
+const leftoverStack = new AsyncDisposableStack();
+afterAll(async () => {
+  try {
+    await leftoverStack.disposeAsync();
+  } catch (e) {
+    console.error('Failed to dispose leftover stack', e);
+  }
+});
+
+it('consumes the pubsub topics and resolves the types correctly', async () => {
+  await using products = await service('products');
+  await using composition = await compose({ output: 'graphql', services: [products] });
+  await using gw = await serve({ supergraph: composition.output, env: redisEnv });
+  const sseClient = createClient({
+    retryAttempts: 0,
+    url: `http://0.0.0.0:${gw.port}/graphql`,
+    fetchFn: fetch,
+  });
+  const iterator = sseClient.iterate({
+    query: /* GraphQL */ `
+      subscription {
+        newProduct {
+          id
+          name
+          price
+        }
+      }
+    `,
+  });
+  leftoverStack.defer(() => iterator.return?.() as any);
+  const pub = new Redis({
+    host: redisEnv.REDIS_HOST,
+    port: redisEnv.REDIS_PORT,
+  });
+  leftoverStack.defer(() => pub.disconnect());
+  for (let i = 0; i < 3; i++) {
+    const id = i + '';
+    const publishing = (async () => {
+      // Publish messages after making sure the user's subscribed
+      await setTimeout(500);
+      await pub.publish('gw:new_product', JSON.stringify({ id }));
+    })();
+    await expect(iterator.next()).resolves.toMatchObject({
+      value: {
+        data: {
+          newProduct: {
+            id,
+            name: 'Roomba X' + id,
+            price: 100,
+          },
+        },
+      },
+      done: false,
+    });
+    // Avait publishing to ensure no unhandled rejections
+    await publishing;
+  }
+});

packages/legacy/config/yaml-config.graphql

@@ -147,6 +147,7 @@ type AdditionalSubscriptionObject {
   pubsubTopic: String!
   result: String
   filterBy: String
+  sourceName: String
 }
 
 union PubSub = String | PubSubConfig

packages/legacy/types/src/config-schema.json

@@ -810,6 +810,9 @@
         },
         "filterBy": {
           "type": "string"
+        },
+        "sourceName": {
+          "type": "string"
         }
       },
       "required": ["targetTypeName", "targetFieldName", "pubsubTopic"]

packages/legacy/types/src/config.ts

@@ -1800,6 +1800,7 @@ export interface AdditionalSubscriptionObject {
   pubsubTopic: string;
   result?: string;
   filterBy?: string;
+  sourceName?: string;
 }
 /**
  * Backend cache

packages/legacy/utils/src/resolve-additional-resolvers.ts

@@ -20,8 +20,15 @@ import {
   type MeshPubSub,
   type YamlConfig,
 } from '@graphql-mesh/types';
-import type { IResolvers, MaybePromise } from '@graphql-tools/utils';
+import {
+  resolveExternalValue,
+  Subschema,
+  type MergedTypeResolver,
+  type StitchingInfo,
+} from '@graphql-tools/delegate';
+import type { IResolvers, Maybe, MaybePromise } from '@graphql-tools/utils';
 import { parseSelectionSet } from '@graphql-tools/utils';
+import { handleMaybePromise } from '@whatwg-node/promise-helpers';
 import { loadFromModuleExportExpression } from './load-from-module-export-expression.js';
 import { withFilter } from './with-filter.js';
 
@@ -202,11 +209,49 @@ export function resolveAdditionalResolversWithoutImport(
       [additionalResolver.targetTypeName]: {
         [additionalResolver.targetFieldName]: {
           subscribe: subscribeFn,
-          resolve: (payload: any) => {
-            if (baseOptions.valuesFromResults) {
-              return baseOptions.valuesFromResults(payload);
+          resolve: (payload: any, _, ctx, info) => {
+            function handlePayload(payload: any) {
+              if (baseOptions.valuesFromResults) {
+                return baseOptions.valuesFromResults(payload);
+              }
+              return payload;
+            }
+            if (additionalResolver.sourceName) {
+              const stitchingInfo = info?.schema.extensions?.stitchingInfo as Maybe<
+                StitchingInfo<any>
+              >;
+              if (!stitchingInfo) {
+                throw new Error(
+                  `Stitching Information object not found in the resolve information, contact maintainers!`,
+                );
+              }
+              const returnTypeName = getNamedType(info.returnType).name;
+              const mergedTypeInfo = stitchingInfo?.mergedTypes?.[returnTypeName];
+              if (!mergedTypeInfo) {
+                throw new Error(
+                  `This "${returnTypeName}" type is not a merged type, disable typeMerging in the config!`,
+                );
+              }
+              const subschema = Array.from(stitchingInfo.subschemaMap?.values() || []).find(
+                s => s.name === additionalResolver.sourceName,
+              );
+              if (!subschema) {
+                throw new Error(`The source "${additionalResolver.sourceName}" is not found`);
+              }
+              const resolver = mergedTypeInfo?.resolvers?.get(subschema);
+              if (!resolver) {
+                throw new Error(
+                  `The type "${returnTypeName}" is not resolvable from the source "${additionalResolver.sourceName}", check your typeMerging configuration!`,
+                );
+              }
+              const selectionSet = info.fieldNodes[0].selectionSet;
+              return handleMaybePromise(
+                () =>
+                  resolver(payload, ctx, info, subschema, selectionSet, undefined, info.returnType),
+                handlePayload,
+              );
             }
-            return payload;
+            return handlePayload(payload);
           },
         },
       },