feat: add PQL manifest support for persisted operations

cdn-server/cdn/src/index.ts

@@ -26,12 +26,12 @@ export interface BlobStorage {
   headObject({
     context,
     key,
-    schemaVersionId,
+    version,
   }: {
     context: Context;
     abortSignal?: AbortSignal;
     key: string;
-    schemaVersionId: string;
+    version: string;
   }): Promise<boolean>;
 }
 
@@ -106,6 +106,9 @@ const jwtMiddleware = (secret: string | ((c: Context) => string)) => {
   };
 };
 
+// Deprecated: Individual persisted operation lookups via CDN are deprecated.
+// The router now downloads all operations at once via the PQL manifest, avoiding
+// per-request latency. This handler is kept for backward compatibility with older routers.
 const persistedOperation = (storage: BlobStorage) => {
   return async (c: Context) => {
     const organizationId = c.get('authenticatedOrganizationId');
@@ -165,7 +168,7 @@ const latestValidRouterConfig = (storage: BlobStorage) => {
     // starts for the first time, and we need to return a config anyway.
     if (body?.version) {
       try {
-        isModified = await storage.headObject({ context: c, key, schemaVersionId: body.version });
+        isModified = await storage.headObject({ context: c, key, version: body.version });
       } catch (e: any) {
         if (e instanceof BlobNotFoundError) {
           return c.notFound();
@@ -262,6 +265,58 @@ const cacheOperations = (storage: BlobStorage) => {
   };
 };
 
+const persistedOperationsManifest = (storage: BlobStorage) => {
+  return async (c: Context) => {
+    const organizationId = c.get('authenticatedOrganizationId');
+    const federatedGraphId = c.get('authenticatedFederatedGraphId');
+
+    if (organizationId !== c.req.param('organization_id') || federatedGraphId !== c.req.param('federated_graph_id')) {
+      return c.text('Bad Request', 400);
+    }
+
+    const key = `${organizationId}/${federatedGraphId}/operations/manifest.json`;
+
+    const body = await c.req.json();
+
+    let isModified = true;
+
+    // Only check if revision is specified otherwise we assume the router
+    // starts for the first time, and we need to return the manifest anyway.
+    if (body?.revision) {
+      try {
+        isModified = await storage.headObject({ context: c, key, version: body.revision });
+      } catch (e: any) {
+        if (e instanceof BlobNotFoundError) {
+          return c.notFound();
+        }
+        throw e;
+      }
+    }
+
+    if (!isModified) {
+      return c.body(null, 304);
+    }
+
+    let blobObject: BlobObject;
+
+    try {
+      blobObject = await storage.getObject({ context: c, key, cacheControl: 'no-cache' });
+    } catch (e: any) {
+      if (e instanceof BlobNotFoundError) {
+        return c.notFound();
+      }
+      throw e;
+    }
+
+    c.header('Content-Type', 'application/json; charset=UTF-8');
+    c.header('Cache-Control', 'no-cache, no-store, must-revalidate');
+
+    return stream(c, async (stream) => {
+      await stream.pipe(blobObject.stream);
+    });
+  };
+};
+
 const subgraphChecks = (storage: BlobStorage) => {
   return async (c: Context) => {
     const organizationId = c.get('authenticatedOrganizationId');
@@ -301,6 +356,11 @@ export const cdn = <E extends Env, S extends Schema = {}, BasePath extends strin
   hono: Hono<E, S, BasePath>,
   opts: CdnOptions,
 ) => {
+  const manifestPath = '/:organization_id/:federated_graph_id/operations/manifest.json';
+  hono
+    .use(manifestPath, jwtMiddleware(opts.authJwtSecret))
+    .post(manifestPath, persistedOperationsManifest(opts.blobStorage));
+
   const operations = '/:organization_id/:federated_graph_id/operations/:client_id/:operation{.+\\.json$}';
   const latestValidRouterConfigs = '/:organization_id/:federated_graph_id/routerconfigs/latest.json';
   hono.use(operations, jwtMiddleware(opts.authJwtSecret)).get(operations, persistedOperation(opts.blobStorage));

cdn-server/cdn/test/cdn.test.ts

@@ -30,12 +30,12 @@ class InMemoryBlobStorage implements BlobStorage {
     return Promise.resolve({ stream, metadata: obj.metadata });
   }
 
-  headObject({ key, schemaVersionId }: { key: string; schemaVersionId: string }): Promise<boolean> {
+  headObject({ key, version }: { key: string; version: string }): Promise<boolean> {
     const obj = this.objects.get(key);
     if (!obj) {
       return Promise.reject(new BlobNotFoundError(`Object with key ${key} not found`));
     }
-    if (obj.metadata?.version === schemaVersionId) {
+    if (obj.metadata?.version === version) {
       return Promise.resolve(false);
     }
     return Promise.resolve(true);
@@ -554,6 +554,182 @@ describe('CDN handlers', () => {
     });
   });
 
+  describe('Test persisted operations manifest handler', async () => {
+    const federatedGraphId = 'federatedGraphId';
+    const organizationId = 'organizationId';
+    const token = await generateToken(organizationId, federatedGraphId, secretKey);
+    const blobStorage = new InMemoryBlobStorage();
+    const requestPath = `/${organizationId}/${federatedGraphId}/operations/manifest.json`;
+
+    const app = new Hono();
+
+    cdn(app, {
+      authJwtSecret: secretKey,
+      authAdmissionJwtSecret: secretAdmissionKey,
+      blobStorage,
+    });
+
+    test('it returns a 401 if no Authorization header is provided', async () => {
+      const res = await app.request(requestPath, {
+        method: 'POST',
+        body: JSON.stringify({}),
+      });
+      expect(res.status).toBe(401);
+    });
+
+    test('it returns a 401 if an invalid Authorization header is provided', async () => {
+      const res = await app.request(requestPath, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token.slice(0, -1)}}`,
+        },
+        body: JSON.stringify({}),
+      });
+      expect(res.status).toBe(401);
+    });
+
+    test('it returns a 400 if the graph or organization ids does not match with the JWT payload', async () => {
+      const res = await app.request(`/foo/bar/operations/manifest.json`, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token}`,
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({}),
+      });
+      expect(res.status).toBe(400);
+    });
+
+    test('it returns a 401 if the token has expired', async () => {
+      const token = await new SignJWT({
+        organization_id: organizationId,
+        federated_graph_id: federatedGraphId,
+        exp: Math.floor(Date.now() / 1000) - 60,
+      })
+        .setProtectedHeader({ alg: 'HS256' })
+        .sign(new TextEncoder().encode(secretKey));
+      const res = await app.request(requestPath, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token}`,
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({}),
+      });
+      expect(res.status).toBe(401);
+    });
+
+    test('it returns the manifest on first request without revision', async () => {
+      const manifestContents = JSON.stringify({
+        version: 1,
+        revision: 'abc123',
+        generatedAt: '2025-01-01T00:00:00.000Z',
+        operations: {
+          sha256hash1: 'query { hello }',
+        },
+      });
+
+      blobStorage.objects.set(`${organizationId}/${federatedGraphId}/operations/manifest.json`, {
+        buffer: Buffer.from(manifestContents),
+        metadata: { version: 'abc123' },
+      });
+
+      const res = await app.request(requestPath, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token}`,
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({}),
+      });
+      expect(res.status).toBe(200);
+      expect(res.headers.get('Content-Type')).toBe('application/json; charset=UTF-8');
+      expect(res.headers.get('Cache-Control')).toBe('no-cache, no-store, must-revalidate');
+      expect(await res.text()).toBe(manifestContents);
+    });
+
+    test('it returns 304 when revision matches the current revision', async () => {
+      blobStorage.objects.set(`${organizationId}/${federatedGraphId}/operations/manifest.json`, {
+        buffer: Buffer.from(JSON.stringify({ version: 1, revision: 'abc123', operations: {} })),
+        metadata: { version: 'abc123' },
+      });
+
+      const res = await app.request(requestPath, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token}`,
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({ revision: 'abc123' }),
+      });
+      expect(res.status).toBe(304);
+    });
+
+    test('it returns 200 when revision does not match the current revision', async () => {
+      const manifestContents = JSON.stringify({
+        version: 1,
+        revision: 'def456',
+        generatedAt: '2025-01-01T00:00:00.000Z',
+        operations: {
+          sha256hash1: 'query { hello }',
+        },
+      });
+
+      blobStorage.objects.set(`${organizationId}/${federatedGraphId}/operations/manifest.json`, {
+        buffer: Buffer.from(manifestContents),
+        metadata: { version: 'def456' },
+      });
+
+      const res = await app.request(requestPath, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token}`,
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({ revision: 'old-revision' }),
+      });
+      expect(res.status).toBe(200);
+      expect(await res.text()).toBe(manifestContents);
+    });
+
+    test('it returns a 404 if the manifest does not exist', async () => {
+      const otherBlobStorage = new InMemoryBlobStorage();
+      const otherApp = new Hono();
+
+      cdn(otherApp, {
+        authJwtSecret: secretKey,
+        authAdmissionJwtSecret: secretAdmissionKey,
+        blobStorage: otherBlobStorage,
+      });
+
+      const res = await otherApp.request(requestPath, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${token}`,
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({}),
+      });
+      expect(res.status).toBe(404);
+    });
+
+    test('it does not conflict with the individual persisted operations route', async () => {
+      const operationContents = JSON.stringify({ version: 1, body: 'query { hello }' });
+      blobStorage.objects.set(`${organizationId}/${federatedGraphId}/operations/clientName/operation.json`, {
+        buffer: Buffer.from(operationContents),
+      });
+
+      const res = await app.request(`/${organizationId}/${federatedGraphId}/operations/clientName/operation.json`, {
+        method: 'GET',
+        headers: {
+          Authorization: `Bearer ${token}`,
+        },
+      });
+      expect(res.status).toBe(200);
+      expect(await res.text()).toBe(operationContents);
+    });
+  });
+
   describe('schema check extensions handler', async () => {
     const organizationId = 'organizationId';
     const checkId = randomUUID();

cdn-server/src/s3.ts

@@ -49,15 +49,7 @@ class S3BlobStorage implements BlobStorage {
     }
   }
 
-  async headObject({
-    context,
-    key,
-    schemaVersionId,
-  }: {
-    context: Context;
-    key: string;
-    schemaVersionId: string;
-  }): Promise<boolean> {
+  async headObject({ context, key, version }: { context: Context; key: string; version: string }): Promise<boolean> {
     const command = new HeadObjectCommand({
       Bucket: this.bucketName,
       Key: key,
@@ -72,7 +64,7 @@ class S3BlobStorage implements BlobStorage {
       } else if (resp.$metadata.httpStatusCode !== 200) {
         throw new Error(`Failed to fetch the metadata of the object.`);
       }
-      if (resp.Metadata && resp.Metadata.version === schemaVersionId) {
+      if (resp.Metadata && resp.Metadata.version === version) {
         return false;
       }
       return true;

controlplane/src/core/bufservices/persisted-operation/deletePersistedOperation.ts

@@ -5,13 +5,12 @@ import type {
   DeletePersistedOperationRequest,
   DeletePersistedOperationResponse,
 } from '@wundergraph/cosmo-connect/dist/platform/v1/platform_pb';
-import type { BlobStorage } from '../../blobstorage/index.js';
 import { FederatedGraphRepository } from '../../repositories/FederatedGraphRepository.js';
 import { UnauthorizedError } from '../../errors/errors.js';
 import { OperationsRepository } from '../../repositories/OperationsRepository.js';
 import type { RouterOptions } from '../../routes.js';
-import type { PersistedOperationWithClientDTO } from '../../../types/index.js';
 import { enrichLogger, getLogger, handleError } from '../../util.js';
+import { generateAndUploadManifest } from './generateManifest.js';
 import { createBlobStoragePath } from './utils.js';
 
 export function deletePersistedOperation(
@@ -76,9 +75,19 @@ export function deletePersistedOperation(
     });
 
     try {
-      await opts.blobStorage.deleteObject({
-        key: path,
-      });
+      await Promise.all([
+        opts.blobStorage.deleteObject({
+          key: path,
+        }),
+        generateAndUploadManifest({
+          db: opts.db,
+          federatedGraphId: federatedGraph.id,
+          organizationId: authContext.organizationId,
+          blobStorage: opts.blobStorage,
+          logger,
+        }),
+      ]);
+
       return {
         response: {
           code: EnumStatusCode.OK,