feat: new execute-subscription-event plugin

.changeset/six-crabs-flash.md

@@ -0,0 +1,5 @@
+---
+'@envelop/execute-subscription-event': patch
+---
+
+initial release

packages/core/test/common.ts

@@ -1,4 +1,5 @@
 import { makeExecutableSchema } from '@graphql-tools/schema';
+import { mapAsyncIterator } from '../src';
 
 export const schema = makeExecutableSchema({
   typeDefs: /* GraphQL */ `
@@ -13,6 +14,7 @@ export const schema = makeExecutableSchema({
 
     type Subscription {
       alphabet: String!
+      message: String!
     }
   `,
   resolvers: {
@@ -21,6 +23,17 @@ export const schema = makeExecutableSchema({
         return { _id: 1, firstName: 'Dotan', lastName: 'Simha' };
       },
     },
+    Subscription: {
+      message: {
+        subscribe: (_, __, context) => {
+          if (!context || 'subscribeSource' in context === false) {
+            throw new Error('No subscribeSource provided for context :(');
+          }
+          return context.subscribeSource;
+        },
+        resolve: (_, __, context) => context.message,
+      },
+    },
     User: {
       id: u => u._id,
       name: u => `${u.firstName} ${u.lastName}`,
@@ -36,3 +49,9 @@ export const query = /* GraphQL */ `
     }
   }
 `;
+
+export const subscriptionOperationString = /* GraphQL */ `
+  subscription {
+    message
+  }
+`;

packages/plugins/execute-subscription-event/README.md

@@ -0,0 +1,51 @@
+## `@envelop/execute-subscription-event`
+
+Utilities for hooking into the [ExecuteSubscriptionEvent](<https://spec.graphql.org/draft/#ExecuteSubscriptionEvent()>) phase.
+
+### `useContextValuePerExecuteSubscriptionEvent`
+
+Create a new context object per `ExecuteSubscriptionEvent` phase, allowing to bypass common issues with context objects such as [`DataLoader`](https://github.com/dotansimha/envelop/issues/80) [caching](https://github.com/graphql/graphql-js/issues/894) [issues](https://github.com/apollographql/subscriptions-transport-ws/issues/330).
+
+```ts
+import { envelop } from '@envelop/core';
+import { useContextValuePerExecuteSubscriptionEvent } from '@envelop/execute-subscription-event';
+import { createContext, createDataLoaders } from './context';
+
+const getEnveloped = envelop({
+  plugins: [
+    useContext(() => createContext())
+    useContextValuePerExecuteSubscriptionEvent(() => ({
+      // Existing context is merged with this context partial
+      // By recreating the DataLoader we ensure no DataLoader caches from the previous event/initial field subscribe call are are hit
+      contextPartial: {
+        dataLoaders: createDataLoaders()
+      },
+    })),
+    // ... other plugins ...
+  ],
+});
+```
+
+Alternatively, you can also provide a callback that is invoked after each [`ExecuteSubscriptionEvent`](<https://spec.graphql.org/draft/#ExecuteSubscriptionEvent()>) phase.
+
+```ts
+import { envelop } from '@envelop/core';
+import { useContextValuePerExecuteSubscriptionEvent } from '@envelop/execute-subscription-event';
+import { createContext, createDataLoaders } from './context';
+
+const getEnveloped = envelop({
+  plugins: [
+    useContext(() => createContext())
+    useContextValuePerExecuteSubscriptionEvent(({ args }) => ({
+      onEnd: () => {
+        // Note that onEnd is invoked only after each ExecuteSubscriptionEvent phase
+        // This means the initial event will still use the cache from potential subscribe dataloader calls
+        // If you use this to clear DataLoader caches it is recommended to not do any DataLoader calls within your field subscribe function.
+        args.contextValue.dataLoaders.users.clearAll()
+        args.contextValue.dataLoaders.posts.clearAll()
+      }
+    })),
+    // ... other plugins ...
+  ],
+});
+```

packages/plugins/execute-subscription-event/package.json

@@ -0,0 +1,49 @@
+{
+  "name": "@envelop/execute-subscription-event",
+  "version": "0.0.0",
+  "author": "Laurin Quast <[email protected]>",
+  "license": "MIT",
+  "sideEffects": false,
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/dotansimha/envelop.git",
+    "directory": "packages/plugins/disable-introspection"
+  },
+  "main": "dist/index.js",
+  "module": "dist/index.mjs",
+  "exports": {
+    ".": {
+      "require": "./dist/index.js",
+      "import": "./dist/index.mjs"
+    },
+    "./*": {
+      "require": "./dist/*.js",
+      "import": "./dist/*.mjs"
+    }
+  },
+  "typings": "dist/index.d.ts",
+  "typescript": {
+    "definition": "dist/index.d.ts"
+  },
+  "scripts": {
+    "test": "jest",
+    "prepack": "bob prepack"
+  },
+  "dependencies": {},
+  "devDependencies": {
+    "@n1ru4l/push-pull-async-iterable-iterator": "3.0.0",
+    "bob-the-bundler": "1.4.1",
+    "graphql": "15.5.1",
+    "typescript": "4.3.5"
+  },
+  "peerDependencies": {
+    "graphql": "^14.0.0 || ^15.0.0"
+  },
+  "buildOptions": {
+    "input": "./src/index.ts"
+  },
+  "publishConfig": {
+    "directory": "dist",
+    "access": "public"
+  }
+}

packages/plugins/execute-subscription-event/src/index.ts

@@ -0,0 +1,42 @@
+import { SubscriptionArgs, execute } from 'graphql';
+import { Plugin } from '@envelop/types';
+import { makeExecute, DefaultContext } from '@envelop/core';
+import { PromiseOrValue } from 'graphql/jsutils/PromiseOrValue';
+import { subscribe } from './subscribe';
+
+export type ContextFactoryOptions = {
+  /** The arguments with which the subscription was set up. */
+  args: SubscriptionArgs;
+};
+
+export type ContextFactoryHook<TContextValue> = {
+  /** Context that will be used for the "ExecuteSubscriptionEvent" phase. */
+  contextPartial: Partial<TContextValue>;
+  /** Optional callback that is invoked once the "ExecuteSubscriptionEvent" phase has ended. Useful for cleanup, such as tearing down database connections. */
+  onEnd?: () => void;
+};
+
+export type ContextFactoryType<TContextValue = DefaultContext> = (
+  options: ContextFactoryOptions
+) => PromiseOrValue<ContextFactoryHook<TContextValue> | void>;
+
+export const useExtendContextValuePerExecuteSubscriptionEvent = <TContextValue = unknown>(
+  createContext: ContextFactoryType<TContextValue>
+): Plugin<TContextValue> => {
+  return {
+    onSubscribe({ args, setSubscribeFn }) {
+      const executeNew = makeExecute(async executionArgs => {
+        const context = await createContext({ args });
+        try {
+          return execute({
+            ...executionArgs,
+            contextValue: { ...executionArgs.contextValue, ...context?.contextPartial },
+          });
+        } finally {
+          context?.onEnd?.();
+        }
+      });
+      setSubscribeFn(subscribe(executeNew));
+    },
+  };
+};

packages/plugins/execute-subscription-event/src/subscribe.ts

@@ -0,0 +1,49 @@
+import { createSourceEventStream } from 'graphql';
+
+import { ExecuteFunction, makeSubscribe, SubscribeFunction } from '@envelop/core';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
+import mapAsyncIterator from 'graphql/subscription/mapAsyncIterator';
+
+/**
+ * This is a almost identical port from graphql-js subscribe.
+ * The only difference is that a custom `execute` function can be injected for customizing the behavior.
+ */
+export const subscribe = (execute: ExecuteFunction): SubscribeFunction =>
+  makeSubscribe(async args => {
+    const { schema, document, rootValue, contextValue, variableValues, operationName, fieldResolver, subscribeFieldResolver } =
+      args;
+
+    const resultOrStream = await createSourceEventStream(
+      schema,
+      document,
+      rootValue,
+      contextValue,
+      variableValues ?? undefined,
+      operationName,
+      subscribeFieldResolver
+    );
+
+    if (!isAsyncIterable(resultOrStream)) {
+      return resultOrStream;
+    }
+
+    // For each payload yielded from a subscription, map it over the normal
+    // GraphQL `execute` function, with `payload` as the rootValue.
+    // This implements the "MapSourceToResponseEvent" algorithm described in
+    // the GraphQL specification. The `execute` function provides the
+    // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
+    // "ExecuteQuery" algorithm, for which `execute` is also used.
+    const mapSourceToResponse = async (payload: object) =>
+      execute({
+        schema,
+        document,
+        rootValue: payload,
+        contextValue,
+        variableValues,
+        operationName,
+        fieldResolver,
+      });
+
+    // Map every source value to a ExecutionResult value as described above.
+    return mapAsyncIterator(resultOrStream, mapSourceToResponse);
+  });

packages/plugins/execute-subscription-event/test/use-extend-context-value-per-subscription-event.spec.ts

@@ -0,0 +1,78 @@
+import { assertStreamExecutionValue, createTestkit } from '@envelop/testing';
+import { schema, subscriptionOperationString } from '../../../core/test/common';
+import { useExtendContextValuePerExecuteSubscriptionEvent } from '../src';
+import { useExtendContext } from '@envelop/core';
+import { makePushPullAsyncIterableIterator } from '@n1ru4l/push-pull-async-iterable-iterator';
+
+describe('useContextValuePerExecuteSubscriptionEvent', () => {
+  it('it can be used for injecting a context that is different from the subscription context', async () => {
+    expect.assertions(4);
+    const { pushValue, asyncIterableIterator } = makePushPullAsyncIterableIterator<unknown>();
+    const subscriptionContextValue: {
+      subscribeSource: AsyncIterableIterator<unknown>;
+      message: string;
+    } = { subscribeSource: asyncIterableIterator, message: 'this is only used during subscribe phase' };
+
+    let counter = 0;
+
+    const testInstance = createTestkit(
+      [
+        useExtendContext(() => subscriptionContextValue),
+        useExtendContextValuePerExecuteSubscriptionEvent(() => ({
+          contextPartial: {
+            message: `${counter}`,
+          },
+        })),
+      ],
+      schema
+    );
+
+    const result = await testInstance.execute(subscriptionOperationString);
+    assertStreamExecutionValue(result);
+
+    pushValue({});
+
+    for await (const value of result) {
+      expect(value.errors).toBeUndefined();
+      if (counter === 0) {
+        expect(value.data?.message).toEqual('0');
+        counter = 1;
+        pushValue({});
+      } else if (counter === 1) {
+        expect(value.data?.message).toEqual('1');
+        return;
+      }
+    }
+  });
+
+  it('invokes cleanup function after value is published', async () => {
+    expect.assertions(3);
+    const { pushValue, asyncIterableIterator } = makePushPullAsyncIterableIterator<unknown>();
+
+    let onEnd = jest.fn();
+    const testInstance = createTestkit(
+      [
+        useExtendContext(() => ({ subscribeSource: asyncIterableIterator })),
+        useExtendContextValuePerExecuteSubscriptionEvent(() => ({
+          contextPartial: {
+            message: `hi`,
+          },
+          onEnd,
+        })),
+      ],
+      schema
+    );
+
+    const result = await testInstance.execute(subscriptionOperationString);
+    assertStreamExecutionValue(result);
+
+    pushValue({});
+
+    for await (const value of result) {
+      expect(value.errors).toBeUndefined();
+      expect(value.data?.message).toEqual('hi');
+      expect(onEnd.mock.calls).toHaveLength(1);
+      return;
+    }
+  });
+});

yarn.lock

@@ -1941,6 +1941,11 @@
     "@n1ru4l/graphql-live-query" "0.7.1"
     "@n1ru4l/push-pull-async-iterable-iterator" "^2.1.4"
 
+"@n1ru4l/[email protected]":
+  version "3.0.0"
+  resolved "https://registry.yarnpkg.com/@n1ru4l/push-pull-async-iterable-iterator/-/push-pull-async-iterable-iterator-3.0.0.tgz#22dc34094c2de5f21b9a798d0ffab16b45de0eb7"
+  integrity sha512-gwoIwo/Dt1GOI+lbcG1G7IeRM2K+Fo0op3OGyFJ4tXUCf2a3Q8lUCm81aoevrXC0nu4gbAXeOWy7wWxjpSvZUw==
+
 "@n1ru4l/push-pull-async-iterable-iterator@^2.1.4":
   version "2.1.4"
   resolved "https://registry.yarnpkg.com/@n1ru4l/push-pull-async-iterable-iterator/-/push-pull-async-iterable-iterator-2.1.4.tgz#a90225474352f9f159bff979905f707b9c6bcf04"