Defer support (#85)

Implement support for the @defer directive. In order to use this feature,
you must be using an appropriate version of `graphql`. At the time of
writing this, @defer is only available in v17 alpha versions of the `graphql`
package, which is currently not officially supported. Due to peer
dependencies, you must install graphql like so in order to force v17:
`npm i graphql@alpha --legacy-peer-deps`

.changeset/honest-bags-check.md

@@ -0,0 +1,6 @@
+---
+'@as-integrations/koa': minor
+---
+
+Implement support for the @defer directive. In order to use this feature, you must be using an appropriate version of `graphql`. At the time of writing this, @defer is only available in v17 alpha versions of the `graphql` package, which is currently not officially supported. Due to peer dependencies, you must install graphql like so in order to force v17:
+`npm i graphql@alpha --legacy-peer-deps`

.circleci/config.yml

@@ -45,6 +45,23 @@ jobs:
       - store_test_results:
           path: junit.xml
 
+  Full incremental delivery tests with graphql-js 17 canary:
+    docker:
+    - image: cimg/base:stable
+    environment:
+      INCREMENTAL_DELIVERY_TESTS_ENABLED: t
+    steps:
+      - setup-node:
+          node-version: "18"
+      # Install a prerelease of graphql-js 17 with incremental delivery support.
+      # --legacy-peer-deps because nothing expects v17 yet.
+      # --no-engine-strict because Node v18 doesn't match the engines fields
+      # on some old stuff.
+      - run: npm i --legacy-peer-deps --no-engine-strict graphql@17.0.0-alpha.1.canary.pr.3361.04ab27334641e170ce0e05bc927b972991953882
+      - run: npm run test:ci
+      - store_test_results:
+          path: junit.xml
+
   Prettier:
     docker:
     - image: cimg/base:stable
@@ -71,5 +88,6 @@ workflows:
                 - "14"
                 - "16"
                 - "18"
+      - Full incremental delivery tests with graphql-js 17 canary
       - Prettier
-      - Spell Check
+      - Spell Check

src/__tests__/integration.test.ts

@@ -11,38 +11,35 @@ import {
 import { koaMiddleware } from '..';
 import { urlForHttpServer } from '../utils';
 
-defineIntegrationTestSuite(
-  async function (
-    serverOptions: ApolloServerOptions<BaseContext>,
-    testOptions?: CreateServerForIntegrationTestsOptions,
-  ) {
-    const app = new Koa();
-    // disable logs to console.error
-    app.silent = true;
+defineIntegrationTestSuite(async function (
+  serverOptions: ApolloServerOptions<BaseContext>,
+  testOptions?: CreateServerForIntegrationTestsOptions,
+) {
+  const app = new Koa();
+  // disable logs to console.error
+  app.silent = true;
 
-    const httpServer = http.createServer(app.callback());
-    const server = new ApolloServer({
-      ...serverOptions,
-      plugins: [
-        ...(serverOptions.plugins ?? []),
-        ApolloServerPluginDrainHttpServer({
-          httpServer,
-        }),
-      ],
-    });
-
-    await server.start();
-    app.use(cors());
-    app.use(bodyParser());
-    app.use(
-      koaMiddleware(server, {
-        context: testOptions?.context,
+  const httpServer = http.createServer(app.callback());
+  const server = new ApolloServer({
+    ...serverOptions,
+    plugins: [
+      ...(serverOptions.plugins ?? []),
+      ApolloServerPluginDrainHttpServer({
+        httpServer,
       }),
-    );
-    await new Promise<void>((resolve) => {
-      httpServer.listen({ port: 0 }, resolve);
-    });
-    return { server, url: urlForHttpServer(httpServer) };
-  },
-  { noIncrementalDelivery: true },
-);
+    ],
+  });
+
+  await server.start();
+  app.use(cors());
+  app.use(bodyParser());
+  app.use(
+    koaMiddleware(server, {
+      context: testOptions?.context,
+    }),
+  );
+  await new Promise<void>((resolve) => {
+    httpServer.listen({ port: 0 }, resolve);
+  });
+  return { server, url: urlForHttpServer(httpServer) };
+});

src/index.ts

@@ -1,11 +1,12 @@
+import { Readable } from 'node:stream';
+import { parse } from 'node:url';
 import type { WithRequired } from '@apollo/utils.withrequired';
 import type {
   ApolloServer,
   BaseContext,
   ContextFunction,
   HTTPGraphQLRequest,
 } from '@apollo/server';
-import { parse } from 'url';
 import type Koa from 'koa';
 // we need the extended `Request` type from `koa-bodyparser`,
 // this is similar to an effectful import but for types, since
@@ -45,7 +46,7 @@ export function koaMiddleware<TContext extends BaseContext>(
   const context: ContextFunction<[KoaContextFunctionArgument], TContext> =
     options?.context ?? defaultContext;
 
-  return async (ctx, next) => {
+  return async ctx => {
     if (!ctx.request.body) {
       // The json koa-bodyparser *always* sets ctx.request.body to {} if it's unset (even
       // if the Content-Type doesn't match), so if it isn't set, you probably
@@ -57,7 +58,7 @@ export function koaMiddleware<TContext extends BaseContext>(
       return;
     }
 
-    const headers = new Map<string, string>();
+    const incomingHeaders = new Map<string, string>();
     for (const [key, value] of Object.entries(ctx.headers)) {
       if (value !== undefined) {
         // Node/Koa headers can be an array or a single value. We join
@@ -66,7 +67,7 @@ export function koaMiddleware<TContext extends BaseContext>(
         // docs on IncomingMessage.headers) and so we don't bother to lower-case
         // them or combine across multiple keys that would lower-case to the
         // same value.
-        headers.set(
+        incomingHeaders.set(
           key,
           Array.isArray(value) ? value.join(', ') : (value as string),
         );
@@ -75,33 +76,40 @@ export function koaMiddleware<TContext extends BaseContext>(
 
     const httpGraphQLRequest: HTTPGraphQLRequest = {
       method: ctx.method.toUpperCase(),
-      headers,
+      headers: incomingHeaders,
       search: parse(ctx.url).search ?? '',
       body: ctx.request.body,
     };
 
-    Object.entries(Object.fromEntries(headers));
+    const { body, headers, status } = await server.executeHTTPGraphQLRequest({
+      httpGraphQLRequest,
+      context: () => context({ ctx }),
+    });
 
-    try {
-      const { body, headers, status } = await server.executeHTTPGraphQLRequest({
-        httpGraphQLRequest,
-        context: () => context({ ctx }),
-      });
-
-      for (const [key, value] of headers) {
-        ctx.set(key, value);
-      }
-
-      ctx.status = status || 200;
-
-      if (body.kind === 'complete') {
-        ctx.body = body.string;
-        return;
-      }
+    if (body.kind === 'complete') {
+      ctx.body = body.string;
+    } else if (body.kind === 'chunked') {
+      ctx.body = Readable.from(async function*() {
+        for await (const chunk of body.asyncIterator) {
+          yield chunk;
+          if (typeof ctx.body.flush === "function") {
+            // If this response has been piped to a writable compression stream then `flush` after
+            // each chunk.
+            // This is identical to the Express integration:
+            // https://github.com/apollographql/apollo-server/blob/a69580565dadad69de701da84092e89d0fddfa00/packages/server/src/express4/index.ts#L96-L105
+            ctx.body.flush();
+          }
+        }
+      }());
+    } else {
+      throw Error(`Delivery method ${(body as any).kind} not implemented`);
+    }
 
-      throw Error('Incremental delivery not implemented');
-    } catch {
-      await next();
+    if (status !== undefined) {
+      ctx.status = status;
+    }
+    for (const [key, value] of headers) {
+      ctx.set(key, value);
     }
   };
 }