Implement server-side rendering support (#261)

* Implement toSuspenseSource for React Suspense

This accepts a Wonka source and throws a promise
if the result is not resolving immediately. This
is done for suspensen and will also limit the
source to a single result.

If the source pushes a value synchronously it
is let through and won't throw a promise, which
also will subsequently let through more values
from the source.

* Add suspense mode to Client

* Implement an SSR exchange

This can be used on the client and the server;
On the server it can start caching results,
while on the client it retrieves cached results
from a rehydrated cache.

* Add tests for SSR exchange

* Fix placement of toSuspenseSource in Client

* Add some smoke tests for SSR

* Add additional client-side rehydration test

* Add SSR data serialization

* Reverse order of cached and forwarded ops in ssrExchange

* Update comments in toSuspenseSource

* Add check to ssr.test.ts to prevent double operation bug

package.json

@@ -117,6 +117,8 @@
     "react": "^16.8.6",
     "react-dom": "^16.8.6",
     "react-hooks-testing-library": "^0.5.1",
+    "react-is": "^16.8.6",
+    "react-ssr-prepass": "^1.0.5",
     "react-test-renderer": "^16.8.6",
     "rimraf": "^2.6.2",
     "terser": "^4.0.0",

src/__snapshots__/client.test.ts.snap

@@ -14,6 +14,7 @@ Client {
   "operations$": [Function],
   "reexecuteOperation": [Function],
   "results$": [Function],
+  "suspense": false,
   "url": "https://hostname.com",
 }
 `;

src/__snapshots__/context.test.ts.snap

@@ -27,6 +27,7 @@ Object {
       "operations$": [Function],
       "reexecuteOperation": [Function],
       "results$": [Function],
+      "suspense": false,
       "url": "/graphql",
     },
     "_currentValue2": Client {
@@ -42,6 +43,7 @@ Object {
       "operations$": [Function],
       "reexecuteOperation": [Function],
       "results$": [Function],
+      "suspense": false,
       "url": "/graphql",
     },
     "_threadCount": 0,
@@ -76,6 +78,7 @@ Object {
       "operations$": [Function],
       "reexecuteOperation": [Function],
       "results$": [Function],
+      "suspense": false,
       "url": "/graphql",
     },
     "_currentValue2": Client {
@@ -91,6 +94,7 @@ Object {
       "operations$": [Function],
       "reexecuteOperation": [Function],
       "results$": [Function],
+      "suspense": false,
       "url": "/graphql",
     },
     "_threadCount": 0,

src/client.ts

@@ -25,6 +25,8 @@ import {
   OperationType,
 } from './types';
 
+import { toSuspenseSource } from './utils';
+
 /** Options for configuring the URQL [client]{@link Client}. */
 export interface ClientOptions {
   /** Target endpoint URL such as `https://my-target:8080/graphql`. */
@@ -33,6 +35,8 @@ export interface ClientOptions {
   fetchOptions?: RequestInit | (() => RequestInit);
   /** An ordered array of Exchanges. */
   exchanges?: Exchange[];
+  /** Activates support for Suspense. */
+  suspense?: boolean;
 }
 
 interface ActiveOperations {
@@ -47,6 +51,7 @@ export class Client {
   url: string;
   fetchOptions?: RequestInit | (() => RequestInit);
   exchange: Exchange;
+  suspense: boolean;
 
   // These are internals to be used to keep track of operations
   dispatchOperation: (operation: Operation) => void;
@@ -57,6 +62,7 @@ export class Client {
   constructor(opts: ClientOptions) {
     this.url = opts.url;
     this.fetchOptions = opts.fetchOptions;
+    this.suspense = !!opts.suspense;
 
     // This subject forms the input of operations; executeOperation may be
     // called to dispatch a new operation on the subject
@@ -128,7 +134,6 @@ export class Client {
   /** Executes an Operation by sending it through the exchange pipeline It returns an observable that emits all related exchange results and keeps track of this observable's subscribers. A teardown signal will be emitted when no subscribers are listening anymore. */
   executeRequestOperation(operation: Operation): Source<OperationResult> {
     const { key, operationName } = operation;
-
     const operationResults$ = pipe(
       this.results$,
       filter(res => res.operation.key === key)
@@ -143,11 +148,13 @@ export class Client {
       );
     }
 
-    return pipe(
+    const result$ = pipe(
       operationResults$,
       onStart<OperationResult>(() => this.onOperationStart(operation)),
       onEnd<OperationResult>(() => this.onOperationEnd(operation))
     );
+
+    return this.suspense ? toSuspenseSource(result$) : result$;
   }
 
   reexecuteOperation = (operation: Operation) => {

src/exchanges/cache.ts

@@ -11,6 +11,9 @@ interface OperationCache {
   [key: string]: Set<number>;
 }
 
+const shouldSkip = ({ operationName }: Operation) =>
+  operationName !== 'mutation' && operationName !== 'query';
+
 export const cacheExchange: Exchange = ({ forward, client }) => {
   const resultCache = new Map() as ResultCache;
   const operationCache = Object.create(null) as OperationCache;
@@ -42,10 +45,6 @@ export const cacheExchange: Exchange = ({ forward, client }) => {
     );
   };
 
-  const shouldSkip = (operation: Operation) =>
-    operation.operationName !== 'mutation' &&
-    operation.operationName !== 'query';
-
   return ops$ => {
     const sharedOps$ = share(ops$);
 

src/exchanges/index.ts

@@ -1,3 +1,4 @@
+export { ssrExchange } from './ssr';
 export { cacheExchange } from './cache';
 export { subscriptionExchange } from './subscription';
 export { debugExchange } from './debug';

src/exchanges/ssr.test.ts

@@ -0,0 +1,91 @@
+import { makeSubject, pipe, map, publish, forEach, Subject } from 'wonka';
+
+import { Client } from '../client';
+import { queryOperation, queryResponse } from '../test-utils';
+import { ExchangeIO, Operation } from '../types';
+import { ssrExchange } from './ssr';
+
+let forward: ExchangeIO;
+let exchangeInput;
+let client: Client;
+let input: Subject<Operation>;
+let output;
+
+beforeEach(() => {
+  input = makeSubject<Operation>();
+  output = jest.fn();
+  forward = ops$ =>
+    pipe(
+      ops$,
+      map(output)
+    );
+  client = { suspense: true } as any;
+  exchangeInput = { forward, client };
+});
+
+it('caches query results correctly', () => {
+  output.mockReturnValueOnce(queryResponse);
+
+  const ssr = ssrExchange();
+  const [ops$, next] = input;
+  const exchange = ssr(exchangeInput)(ops$);
+
+  publish(exchange);
+  next(queryOperation);
+
+  const data = ssr.extractData();
+  expect(Object.keys(data)).toEqual(['' + queryOperation.key]);
+
+  expect(data).toEqual({
+    [queryOperation.key]: {
+      data: queryResponse.data,
+      error: undefined,
+    },
+  });
+});
+
+it('resolves cached query results correctly', () => {
+  const onPush = jest.fn();
+
+  const ssr = ssrExchange({
+    initialState: { [queryOperation.key]: queryResponse as any },
+  });
+
+  const [ops$, next] = input;
+  const exchange = ssr(exchangeInput)(ops$);
+
+  pipe(
+    exchange,
+    forEach(onPush)
+  );
+  next(queryOperation);
+
+  const data = ssr.extractData();
+  expect(Object.keys(data).length).toBe(1);
+  expect(output).not.toHaveBeenCalled();
+  expect(onPush).toHaveBeenCalledWith(queryResponse);
+});
+
+it('deletes cached results in non-suspense environments', () => {
+  client.suspense = false;
+  const onPush = jest.fn();
+  const ssr = ssrExchange();
+
+  ssr.restoreData({ [queryOperation.key]: queryResponse as any });
+  expect(Object.keys(ssr.extractData()).length).toBe(1);
+
+  const [ops$, next] = input;
+  const exchange = ssr(exchangeInput)(ops$);
+
+  pipe(
+    exchange,
+    forEach(onPush)
+  );
+  next(queryOperation);
+
+  expect(Object.keys(ssr.extractData()).length).toBe(0);
+  expect(onPush).toHaveBeenCalledWith(queryResponse);
+
+  // NOTE: The operation should not be duplicated
+  expect(output).not.toHaveBeenCalled();
+});

src/exchanges/ssr.ts

@@ -0,0 +1,127 @@
+import { pipe, share, filter, merge, map, tap } from 'wonka';
+import { Exchange, OperationResult, Operation } from '../types';
+import { CombinedError } from '../utils';
+
+export interface SerializedResult {
+  data?: any;
+  error?: {
+    networkError?: string;
+    graphQLErrors: string[];
+  };
+}
+
+export interface SSRData {
+  [key: string]: SerializedResult;
+}
+
+export interface SSRExchangeParams {
+  initialState?: SSRData;
+}
+
+export interface SSRExchange extends Exchange {
+  /** Rehydrates cached data */
+  restoreData(data: SSRData): void;
+  /** Extracts cached data */
+  extractData(): SSRData;
+}
+
+const shouldSkip = ({ operationName }: Operation) =>
+  operationName !== 'subscription' && operationName !== 'query';
+
+/** Serialize an OperationResult to plain JSON */
+const serializeResult = ({
+  data,
+  error,
+}: OperationResult): SerializedResult => {
+  const result: SerializedResult = { data, error: undefined };
+  if (error !== undefined) {
+    result.error = {
+      networkError: '' + error.networkError,
+      graphQLErrors: error.graphQLErrors.map(x => '' + x),
+    };
+  }
+
+  return result;
+};
+
+/** Deserialize plain JSON to an OperationResult */
+const deserializeResult = (
+  operation: Operation,
+  result: SerializedResult
+): OperationResult => {
+  const { error, data } = result;
+  const deserialized: OperationResult = { operation, data, error: undefined };
+  if (error !== undefined) {
+    deserialized.error = new CombinedError({
+      networkError: new Error(error.networkError),
+      graphQLErrors: error.graphQLErrors,
+    });
+  }
+
+  return deserialized;
+};
+
+/** The ssrExchange can be created to capture data during SSR and also to rehydrate it on the client */
+export const ssrExchange = (params?: SSRExchangeParams): SSRExchange => {
+  const data: SSRData = {};
+
+  const isCached = (operation: Operation) => {
+    return !shouldSkip(operation) && data[operation.key] !== undefined;
+  };
+
+  // The SSR Exchange is a temporary cache that can populate results into data for suspense
+  // On the client it can be used to retrieve these temporary results from a rehydrated cache
+  const ssr: SSRExchange = ({ client, forward }) => ops$ => {
+    const sharedOps$ = share(ops$);
+
+    let forwardedOps$ = pipe(
+      sharedOps$,
+      filter(op => !isCached(op)),
+      forward
+    );
+
+    // NOTE: Since below we might delete the cached entry after accessing
+    // it once, cachedOps$ needs to be merged after forwardedOps$
+    let cachedOps$ = pipe(
+      sharedOps$,
+      filter(op => isCached(op)),
+      map(op => {
+        const serialized = data[op.key];
+        return deserializeResult(op, serialized);
+      })
+    );
+
+    if (client.suspense) {
+      // Inside suspense-mode we cache results in the cache as they're resolved
+      forwardedOps$ = pipe(
+        forwardedOps$,
+        tap((result: OperationResult) => {
+          const { operation } = result;
+          if (!shouldSkip(operation)) {
+            const serialized = serializeResult(result);
+            data[operation.key] = serialized;
+          }
+        })
+      );
+    } else {
+      // Outside of suspense-mode we delete results from the cache as they're resolved
+      cachedOps$ = pipe(
+        cachedOps$,
+        tap((result: OperationResult) => {
+          delete data[result.operation.key];
+        })
+      );
+    }
+
+    return merge([forwardedOps$, cachedOps$]);
+  };
+
+  ssr.restoreData = (restore: SSRData) => Object.assign(data, restore);
+  ssr.extractData = () => Object.assign({}, data);
+
+  if (params && params.initialState) {
+    ssr.restoreData(params.initialState);
+  }
+
+  return ssr;
+};