feat(response-cache): support defer and stream (#1896)

* feat(response-cache): support defer and stream

* chore(dependencies): updated changesets for modified dependencies

* avoid to add undefined errors and extensions keys

* add changeset

* fix types

* remove unused imports

* use @graphql-tools/utils incremental merging

* dicriminate union using 'in' keyword (wtf)

* try graphql tools alpha incremental merging

* use new graphql-tools version

* remove dependency addition from changeset

* chore(dependencies): updated changesets for modified dependencies

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.changeset/@envelop_response-cache-1896-dependencies.md

@@ -0,0 +1,9 @@
+---
+'@envelop/response-cache': patch
+---
+
+dependencies updates:
+
+- Updated dependency
+  [`@graphql-tools/utils@^10.0.3` ↗︎](https://www.npmjs.com/package/@graphql-tools/utils/v/10.0.3)
+  (from `^10.0.0`, in `dependencies`)

.changeset/six-news-chew.md

@@ -0,0 +1,6 @@
+---
+'@envelop/response-cache': minor
+'@envelop/types': patch
+---
+
+add support for @defer and @stream

packages/plugins/response-cache/package.json

@@ -51,14 +51,15 @@
     "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0"
   },
   "dependencies": {
-    "@graphql-tools/utils": "^10.0.0",
+    "@graphql-tools/utils": "^10.0.3",
     "@whatwg-node/fetch": "^0.9.0",
     "fast-json-stable-stringify": "^2.1.0",
     "lru-cache": "^10.0.0",
     "tslib": "^2.5.0"
   },
   "devDependencies": {
     "@envelop/core": "^4.0.0",
+    "@graphql-tools/executor": "^1.1.0",
     "@graphql-tools/schema": "10.0.0",
     "graphql": "16.6.0",
     "ioredis-mock": "5.9.1",

packages/plugins/response-cache/src/plugin.ts

@@ -18,7 +18,14 @@ import {
   ObjMap,
   Plugin,
 } from '@envelop/core';
-import { getDirective, MapperKind, mapSchema, memoize1, visitResult } from '@graphql-tools/utils';
+import {
+  getDirective,
+  MapperKind,
+  mapSchema,
+  memoize1,
+  mergeIncrementalResult,
+  visitResult,
+} from '@graphql-tools/utils';
 import type { Cache, CacheEntityRecord } from './cache.js';
 import { hashSHA256 } from './hash-sha256.js';
 import { createInMemoryCache } from './in-memory-cache.js';
@@ -463,42 +470,82 @@ export function useResponseCache<PluginContext extends Record<string, any> = {}>
         );
       }
 
-      return {
-        onExecuteDone({ result, setResult }) {
-          if (isAsyncIterable(result)) {
-            // eslint-disable-next-line no-console
-            console.warn(
-              '[useResponseCache] AsyncIterable returned from execute is currently unsupported.',
-            );
-            return;
-          }
+      async function maybeCacheResult(
+        result: ExecutionResult,
+        setResult: (newResult: ExecutionResult) => void,
+      ) {
+        const processedResult = processResult(result) as ResponseCacheExecutionResult;
 
-          const processedResult = processResult(result) as ResponseCacheExecutionResult;
+        if (skip) {
+          return;
+        }
 
-          if (skip) {
-            return;
-          }
+        if (!shouldCacheResult({ cacheKey, result: processedResult })) {
+          return;
+        }
 
-          if (!shouldCacheResult({ cacheKey, result: processedResult })) {
-            return;
+        // we only use the global ttl if no currentTtl has been determined.
+        const finalTtl = currentTtl ?? globalTtl;
+
+        if (finalTtl === 0) {
+          if (includeExtensionMetadata) {
+            setResult(resultWithMetadata(processedResult, { hit: false, didCache: false }));
           }
+          return;
+        }
 
-          // we only use the global ttl if no currentTtl has been determined.
-          const finalTtl = currentTtl ?? globalTtl;
+        cache.set(cacheKey, processedResult, identifier.values(), finalTtl);
+        if (includeExtensionMetadata) {
+          setResult(
+            resultWithMetadata(processedResult, { hit: false, didCache: true, ttl: finalTtl }),
+          );
+        }
+      }
 
-          if (finalTtl === 0) {
-            if (includeExtensionMetadata) {
-              setResult(resultWithMetadata(processedResult, { hit: false, didCache: false }));
-            }
+      return {
+        onExecuteDone(payload) {
+          if (!isAsyncIterable(payload.result)) {
+            maybeCacheResult(payload.result, payload.setResult);
             return;
           }
 
-          cache.set(cacheKey, processedResult, identifier.values(), finalTtl);
-          if (includeExtensionMetadata) {
-            setResult(
-              resultWithMetadata(processedResult, { hit: false, didCache: true, ttl: finalTtl }),
-            );
-          }
+          // When the result is an AsyncIterable, it means the query is using @defer or @stream.
+          // This means we have to build the final result by merging the incremental results.
+          // The merged result is then used to know if we should cache it and to calculate the ttl.
+          let result: ExecutionResult = {};
+          return {
+            onNext(payload) {
+              const { data, errors, extensions } = payload.result;
+
+              if (data) {
+                // This is the first result with the initial data payload sent to the client. We use it as the base result
+                if (data) {
+                  result = { data };
+                }
+                if (errors) {
+                  result.errors = errors;
+                }
+                if (extensions) {
+                  result.extensions = extensions;
+                }
+              }
+
+              if ('hasNext' in payload.result) {
+                const { incremental, hasNext } = payload.result;
+
+                if (incremental) {
+                  for (const patch of incremental) {
+                    mergeIncrementalResult({ executionResult: result, incrementalResult: patch });
+                  }
+                }
+
+                if (!hasNext) {
+                  // The query is complete, we can process the final result
+                  maybeCacheResult(result, payload.setResult);
+                }
+              }
+            },
+          };
         },
       };
     },

packages/plugins/response-cache/test/response-cache.spec.ts

@@ -1,10 +1,13 @@
 import { getIntrospectionQuery, GraphQLObjectType, GraphQLSchema } from 'graphql';
-import { useLogger } from '@envelop/core';
+import * as GraphQLJS from 'graphql';
+import { envelop, useEngine, useLogger, useSchema } from '@envelop/core';
 import { useGraphQlJit } from '@envelop/graphql-jit';
 import { useParserCache } from '@envelop/parser-cache';
 import { assertSingleExecutionValue, createTestkit, TestkitInstance } from '@envelop/testing';
 import { useValidationCache } from '@envelop/validation-cache';
+import { normalizedExecutor } from '@graphql-tools/executor';
 import { makeExecutableSchema } from '@graphql-tools/schema';
+import { mapSchema as cloneSchema } from '@graphql-tools/utils';
 import {
   cacheControlDirective,
   createInMemoryCache,
@@ -2822,4 +2825,184 @@ describe('useResponseCache', () => {
       expect(spy).toHaveBeenCalledTimes(2);
     });
   });
+
+  it('should cache queries using @stream', async () => {
+    const spy = jest.fn(async function* () {
+      yield {
+        id: 1,
+        name: 'User 1',
+        comments: [{ id: 1, text: 'Comment 1 of User 1' }],
+      };
+      yield { id: 2, name: 'User 2', comments: [] };
+      await new Promise(process.nextTick);
+      yield { id: 3, name: 'User 3', comments: [] };
+    });
+    const schema = makeExecutableSchema({
+      typeDefs: /* GraphQL */ `
+        directive @stream on FIELD
+
+        type Query {
+          users: [User!]!
+        }
+
+        type Mutation {
+          updateUser(id: ID!): User!
+        }
+
+        type User {
+          id: ID!
+          name: String!
+          comments: [Comment!]!
+          recentComment: Comment
+        }
+
+        type Comment {
+          id: ID!
+          text: String!
+        }
+      `,
+      resolvers: {
+        Query: {
+          users: spy,
+        },
+      },
+    });
+
+    const testInstance = createTestkit(
+      envelop({
+        plugins: [
+          useEngine({ ...GraphQLJS, execute: normalizedExecutor, subscribe: normalizedExecutor }),
+          useSchema(cloneSchema(schema)),
+          useResponseCache({ session: () => null }),
+        ],
+      }),
+    );
+
+    const query = /* GraphQL */ `
+      query test {
+        users @stream {
+          id
+          name
+          comments {
+            id
+            text
+          }
+        }
+      }
+    `;
+
+    await waitForResult(testInstance.execute(query));
+    expect(await waitForResult(testInstance.execute(query))).toEqual({
+      data: {
+        users: [
+          {
+            id: '1',
+            name: 'User 1',
+            comments: [{ id: '1', text: 'Comment 1 of User 1' }],
+          },
+          { id: '2', name: 'User 2', comments: [] },
+          { id: '3', name: 'User 3', comments: [] },
+        ],
+      },
+    });
+    expect(spy).toHaveBeenCalledTimes(1);
+  });
+
+  it('should cache queries using @defer', async () => {
+    const spy = jest.fn(async function* () {
+      yield {
+        id: 1,
+        name: 'User 1',
+        comments: [{ id: 1, text: 'Comment 1 of User 1' }],
+      };
+      yield { id: 2, name: 'User 2', comments: [] };
+      await new Promise(process.nextTick);
+      yield { id: 3, name: 'User 3', comments: [] };
+    });
+    const schema = makeExecutableSchema({
+      typeDefs: /* GraphQL */ `
+        directive @defer on FRAGMENT_SPREAD | INLINE_FRAGMENT
+
+        type Query {
+          users: [User!]!
+        }
+
+        type Mutation {
+          updateUser(id: ID!): User!
+        }
+
+        type User {
+          id: ID!
+          name: String!
+          comments: [Comment!]!
+          recentComment: Comment
+        }
+
+        type Comment {
+          id: ID!
+          text: String!
+        }
+      `,
+      resolvers: {
+        Query: {
+          users: spy,
+        },
+      },
+    });
+
+    const testInstance = createTestkit(
+      envelop({
+        plugins: [
+          useEngine({ ...GraphQLJS, execute: normalizedExecutor, subscribe: normalizedExecutor }),
+          useSchema(cloneSchema(schema)),
+          useResponseCache({ session: () => null, includeExtensionMetadata: true }),
+        ],
+      }),
+    );
+
+    const query = /* GraphQL */ `
+      query test {
+        users {
+          id
+          name
+
+          ... on User @defer {
+            comments {
+              id
+              text
+            }
+          }
+        }
+      }
+    `;
+
+    await waitForResult(testInstance.execute(query));
+    expect(await waitForResult(testInstance.execute(query))).toEqual({
+      data: {
+        users: [
+          {
+            id: '1',
+            name: 'User 1',
+            comments: [{ __typename: 'Comment', id: '1', text: 'Comment 1 of User 1' }],
+          },
+          { id: '2', name: 'User 2', comments: [] },
+          { id: '3', name: 'User 3', comments: [] },
+        ],
+      },
+      extensions: { responseCache: { hit: true } },
+    });
+    expect(spy).toHaveBeenCalledTimes(1);
+  });
 });
+
+async function waitForResult(result: any) {
+  result = await result;
+  if (result.next) {
+    let res = [];
+    for await (const r of result) {
+      res.push(r);
+    }
+  }
+
+  return result;
+}

packages/types/src/graphql.ts

@@ -79,3 +79,36 @@ export interface ExecutionResult<TData = ObjMap<unknown>, TExtensions = ObjMap<u
   data?: TData | null;
   extensions?: TExtensions;
 }
+
+export interface IncrementalDeferResult<
+  TData = Record<string, unknown>,
+  TExtensions = Record<string, unknown>,
+> extends ExecutionResult<TData, TExtensions> {
+  path?: ReadonlyArray<string | number>;
+  label?: string;
+}
+
+export interface IncrementalStreamResult<
+  TData = Array<unknown>,
+  TExtensions = Record<string, unknown>,
+> {
+  errors?: ReadonlyArray<any>;
+  items?: TData | null;
+  path?: ReadonlyArray<string | number>;
+  label?: string;
+  extensions?: TExtensions;
+}
+
+export type IncrementalResult<
+  TData = Record<string, unknown>,
+  TExtensions = Record<string, unknown>,
+> = IncrementalDeferResult<TData, TExtensions> | IncrementalStreamResult<TData, TExtensions>;
+
+export interface IncrementalExecutionResult<
+  TData = Record<string, unknown>,
+  TExtensions = Record<string, unknown>,
+> extends ExecutionResult<TData, TExtensions> {
+  hasNext: boolean;
+  incremental?: ReadonlyArray<IncrementalResult<TData, TExtensions>>;
+  extensions?: TExtensions;
+}