feat(gateway): allow pinning the gateway to a specific version of the schema

Author: adambenhassen

integration-tests/tests/api/artifacts-cdn.spec.ts

@@ -10,7 +10,7 @@ import {
   PutObjectCommand,
   S3Client,
 } from '@aws-sdk/client-s3';
-import { createSupergraphManager } from '@graphql-hive/apollo';
+import { createServicesFetcher, createSupergraphManager } from '@graphql-hive/apollo';
 import { graphql } from '../../testkit/gql';
 import { execute } from '../../testkit/graphql';
 import { initSeed } from '../../testkit/seed';
@@ -460,6 +460,103 @@ function runArtifactsCDNTests(
       }
     });
 
+    test.concurrent(
+      'createSupergraphManager with schemaVersionId pins to specific version',
+      async ({ expect }) => {
+        const endpointBaseUrl = await getBaseEndpoint();
+        const { createOrg } = await initSeed().createOwner();
+        const { createProject } = await createOrg();
+        const { createTargetAccessToken, createCdnAccess, target } = await createProject(
+          ProjectType.Federation,
+        );
+        const writeToken = await createTargetAccessToken({});
+
+        // Publish V1 Schema
+        await writeToken
+          .publishSchema({
+            author: 'Kamil',
+            commit: 'v1',
+            sdl: `type Query { ping: String }`,
+            service: 'ping',
+            url: 'http://ping.com',
+          })
+          .then(r => r.expectNoGraphQLErrors());
+
+        const cdnAccessResult = await createCdnAccess();
+
+        // Create manager without pinning to get initial version
+        const manager = createSupergraphManager({
+          endpoint: endpointBaseUrl + target.id,
+          key: cdnAccessResult.secretAccessToken,
+        });
+
+        const gateway = new ApolloGateway({ supergraphSdl: manager });
+        const server = new ApolloServer({ gateway });
+
+        try {
+          await startStandaloneServer(server);
+
+          // Capture the version ID
+          const v1VersionId = manager.getSchemaVersionId();
+          expect(v1VersionId).toBeDefined();
+
+          await server.stop();
+
+          // Publish V2 Schema with different content
+          await writeToken
+            .publishSchema({
+              author: 'Kamil',
+              commit: 'v2',
+              sdl: `type Query { ping: String, pong: String }`,
+              service: 'ping',
+              url: 'http://ping.com',
+            })
+            .then(r => r.expectNoGraphQLErrors());
+
+          // Create a new manager pinned to V1
+          const pinnedManager = createSupergraphManager({
+            endpoint: endpointBaseUrl + target.id,
+            key: cdnAccessResult.secretAccessToken,
+            schemaVersionId: v1VersionId!,
+          });
+
+          const pinnedGateway = new ApolloGateway({ supergraphSdl: pinnedManager });
+          const pinnedServer = new ApolloServer({ gateway: pinnedGateway });
+
+          try {
+            const { url } = await startStandaloneServer(pinnedServer);
+
+            // Query the schema - should only have 'ping', not 'pong'
+            const response = await fetch(url, {
+              method: 'POST',
+              headers: {
+                accept: 'application/json',
+                'content-type': 'application/json',
+              },
+              body: JSON.stringify({
+                query: `{ __schema { types { name fields { name } } } }`,
+              }),
+            });
+
+            expect(response.status).toBe(200);
+            const result = await response.json();
+            const queryType = result.data.__schema.types.find(
+              (t: { name: string }) => t.name === 'Query',
+            );
+            expect(queryType.fields).toContainEqual({ name: 'ping' });
+            expect(queryType.fields).not.toContainEqual({ name: 'pong' });
+
+            // Verify the pinned manager returns V1 version ID
+            expect(pinnedManager.getSchemaVersionId()).toBe(v1VersionId);
+          } finally {
+            await pinnedServer.stop();
+          }
+        } finally {
+          // Server already stopped in try block
+        }
+      },
+    );
+
     test.concurrent('access versioned SDL artifact with valid credentials', async ({ expect }) => {
       const { createOrg } = await initSeed().createOwner();
       const { createProject } = await createOrg();
@@ -1230,6 +1327,78 @@ function runArtifactsCDNTests(
         expect(contractSupergraphResponse.headers.get('x-hive-schema-version-id')).toBe(versionId);
       },
     );
+
+    test.concurrent(
+      'createServicesFetcher with schemaVersionId fetches pinned version',
+      async ({ expect }) => {
+        const endpointBaseUrl = await getBaseEndpoint();
+        const { createOrg } = await initSeed().createOwner();
+        const { createProject } = await createOrg();
+        const { createTargetAccessToken, createCdnAccess, target } = await createProject(
+          ProjectType.Federation,
+        );
+        const writeToken = await createTargetAccessToken({});
+
+        // Publish V1 Schema
+        await writeToken
+          .publishSchema({
+            author: 'Kamil',
+            commit: 'v1',
+            sdl: `type Query { ping: String }`,
+            service: 'ping',
+            url: 'http://ping.com',
+          })
+          .then(r => r.expectNoGraphQLErrors());
+
+        // Get V1 version ID
+        const v1Version = await writeToken.fetchLatestValidSchema();
+        const v1VersionId = v1Version.latestValidVersion?.id;
+        expect(v1VersionId).toBeDefined();
+
+        const cdnAccessResult = await createCdnAccess();
+
+        // Fetch V1 and capture content
+        const v1Fetcher = createServicesFetcher({
+          endpoint: endpointBaseUrl + target.id,
+          key: cdnAccessResult.secretAccessToken,
+        });
+        const v1Result = await v1Fetcher();
+        expect(v1Result[0].schemaVersionId).toBe(v1VersionId);
+
+        // Publish V2 Schema with different content
+        await writeToken
+          .publishSchema({
+            author: 'Kamil',
+            commit: 'v2',
+            sdl: `type Query { ping: String, pong: String }`,
+            service: 'ping',
+            url: 'http://ping.com',
+          })
+          .then(r => r.expectNoGraphQLErrors());
+
+        // Create a pinned fetcher for V1
+        const pinnedFetcher = createServicesFetcher({
+          endpoint: endpointBaseUrl + target.id,
+          key: cdnAccessResult.secretAccessToken,
+          schemaVersionId: v1VersionId!,
+        });
+
+        const pinnedResult = await pinnedFetcher();
+
+        // Should still return V1 content even after V2 was published
+        expect(pinnedResult[0].sdl).toBe(v1Result[0].sdl);
+        expect(pinnedResult[0].sdl).not.toContain('pong');
+
+        // Latest fetcher should return V2
+        const latestFetcher = createServicesFetcher({
+          endpoint: endpointBaseUrl + target.id,
+          key: cdnAccessResult.secretAccessToken,
+        });
+        const latestResult = await latestFetcher();
+        expect(latestResult[0].sdl).toContain('pong');
+        expect(latestResult[0].schemaVersionId).not.toBe(v1VersionId);
+      },
+    );
   });
 }
 

packages/libraries/apollo/src/index.ts

@@ -55,16 +55,30 @@ export type CreateSupergraphManagerArgs = {
    * Default: currents package version
    */
   version?: string;
+  /**
+   * Pin to a specific schema version.
+   * When provided, fetches from the versioned endpoint
+   */
+  schemaVersionId?: string;
 };
 
 export function createSupergraphManager(args: CreateSupergraphManagerArgs) {
   const logger = args.logger ?? new Logger({ level: false });
   const pollIntervalInMs = args.pollIntervalInMs ?? 30_000;
   let endpoints = Array.isArray(args.endpoint) ? args.endpoint : [args.endpoint];
 
-  const endpoint = endpoints.map(endpoint =>
-    endpoint.endsWith('/supergraph') ? endpoint : joinUrl(endpoint, 'supergraph'),
-  );
+  // Build endpoints - use versioned path if schemaVersionId is provided
+  const endpoint = endpoints.map(ep => {
+    // Remove trailing /supergraph if present to get the base
+    const base = ep.endsWith('/supergraph') ? ep.slice(0, -'/supergraph'.length) : ep;
+
+    if (args.schemaVersionId) {
+      // Versioned endpoint: /artifacts/v1/{targetId}/version/{versionId}/supergraph
+      return joinUrl(joinUrl(joinUrl(base, 'version'), args.schemaVersionId), 'supergraph');
+    }
+
+    return joinUrl(base, 'supergraph');
+  });
 
   const artifactsFetcher = createCDNArtifactFetcher({
     endpoint: endpoint as [string, string],
@@ -79,19 +93,23 @@ export function createSupergraphManager(args: CreateSupergraphManagerArgs) {
   });
 
   let timer: ReturnType<typeof setTimeout> | null = null;
+  let currentSchemaVersionId: string | undefined = args.schemaVersionId;
 
   return {
     async initialize(hooks: { update(supergraphSdl: string): void }): Promise<{
       supergraphSdl: string;
+      schemaVersionId?: string;
       cleanup?: () => Promise<void>;
     }> {
       const initialResult = await artifactsFetcher.fetch();
+      currentSchemaVersionId = initialResult.schemaVersionId ?? args.schemaVersionId;
 
       function poll() {
         timer = setTimeout(async () => {
           try {
             const result = await artifactsFetcher.fetch();
             if (result.contents) {
+              currentSchemaVersionId = result.schemaVersionId ?? undefined;
               hooks.update?.(result.contents);
             }
           } catch (error) {
@@ -105,6 +123,7 @@ export function createSupergraphManager(args: CreateSupergraphManagerArgs) {
 
       return {
         supergraphSdl: initialResult.contents,
+        schemaVersionId: currentSchemaVersionId,
         cleanup: async () => {
           if (timer) {
             clearTimeout(timer);
@@ -113,6 +132,9 @@ export function createSupergraphManager(args: CreateSupergraphManagerArgs) {
         },
       };
     },
+    getSchemaVersionId(): string | undefined {
+      return currentSchemaVersionId;
+    },
   };
 }
 

packages/libraries/core/src/client/gateways.ts

@@ -9,19 +9,46 @@ interface Schema {
   name: string;
 }
 
+function buildServicesEndpoint(baseEndpoint: string, schemaVersionId?: string): string {
+  // Remove trailing /services if present to get the base
+  const base = baseEndpoint.endsWith('/services')
+    ? baseEndpoint.slice(0, -'/services'.length)
+    : baseEndpoint;
+
+  if (schemaVersionId) {
+    // Versioned endpoint: /artifacts/v1/{targetId}/version/{versionId}/services
+    return joinUrl(joinUrl(joinUrl(base, 'version'), schemaVersionId), 'services');
+  }
+
+  // Latest endpoint: /artifacts/v1/{targetId}/services
+  return joinUrl(base, 'services');
+}
+
 function createFetcher(options: SchemaFetcherOptions & ServicesFetcherOptions) {
+  if (options.schemaVersionId !== undefined) {
+    const trimmed = options.schemaVersionId.trim();
+    if (trimmed.length === 0) {
+      throw new Error(
+        'Invalid schemaVersionId: cannot be empty or whitespace. Provide a valid version ID or omit the option to fetch the latest version.',
+      );
+    }
+  }
+
   const logger = chooseLogger(options.logger ?? console);
   let cacheETag: string | null = null;
   let cached: {
-    id: string;
+    schemas: readonly Schema[] | Schema;
     supergraphSdl: string;
+    schemaVersionId: string | null;
   } | null = null;
 
-  const endpoint = options.endpoint.endsWith('/services')
-    ? options.endpoint
-    : joinUrl(options.endpoint, 'services');
+  const endpoint = buildServicesEndpoint(options.endpoint, options.schemaVersionId);
 
-  return function fetcher(): Promise<readonly Schema[] | Schema> {
+  return function fetcher(): Promise<{
+    schemas: readonly Schema[] | Schema;
+    supergraphSdl: string;
+    schemaVersionId: string | null;
+  }> {
     const headers: {
       [key: string]: string;
     } = {
@@ -48,14 +75,23 @@ function createFetcher(options: SchemaFetcherOptions & ServicesFetcherOptions) {
       .then(async response => {
         if (response.ok) {
           const result = await response.json();
+          const schemaVersionId = response.headers.get('x-hive-schema-version-id');
+
+          if (!schemaVersionId) {
+            logger.info(
+              `Response from ${endpoint} did not include x-hive-schema-version-id header. Version pinning will not be available.`,
+            );
+          }
+
+          const data = { schemas: result, schemaVersionId };
 
           const etag = response.headers.get('etag');
           if (etag) {
-            cached = result;
+            cached = data;
             cacheETag = etag;
           }
 
-          return result;
+          return data;
         }
 
         if (response.status === 304 && cached !== null) {
@@ -71,24 +107,26 @@ export function createSchemaFetcher(options: SchemaFetcherOptions) {
   const fetcher = createFetcher(options);
 
   return function schemaFetcher() {
-    return fetcher().then(schema => {
+    return fetcher().then(data => {
+      const { schemas, schemaVersionId } = data;
       let service: Schema;
       // Before the new artifacts endpoint the body returned an array or a single object depending on the project type.
       // This handles both in a backwards-compatible way.
-      if (schema instanceof Array) {
-        if (schema.length !== 1) {
+      if (schemas instanceof Array) {
+        if (schemas.length !== 1) {
           throw new Error(
             'Encountered multiple services instead of a single service. Please use createServicesFetcher instead.',
           );
         }
-        service = schema[0];
+        service = schemas[0];
       } else {
-        service = schema;
+        service = schemas;
       }
 
       return {
         id: createSchemaId(service),
         ...service,
+        ...(schemaVersionId ? { schemaVersionId } : {}),
       };
     });
   };
@@ -98,10 +136,17 @@ export function createServicesFetcher(options: ServicesFetcherOptions) {
   const fetcher = createFetcher(options);
 
   return function schemaFetcher() {
-    return fetcher().then(async services => {
-      if (services instanceof Array) {
+    return fetcher().then(async data => {
+      const { schemas, schemaVersionId } = data;
+      if (schemas instanceof Array) {
         return Promise.all(
-          services.map(service => createSchemaId(service).then(id => ({ id, ...service }))),
+          schemas.map(service =>
+            createSchemaId(service).then(id => ({
+              id,
+              ...service,
+              ...(schemaVersionId ? { schemaVersionId } : {}),
+            })),
+          ),
         );
       }
       throw new Error(