persistant layer for graphcache

src/cacheExchange.ts

@@ -8,7 +8,22 @@ import {
 } from 'urql/core';
 
 import { IntrospectionQuery } from 'graphql';
-import { filter, map, merge, pipe, share, tap } from 'wonka';
+import {
+  filter,
+  map,
+  merge,
+  pipe,
+  share,
+  tap,
+  fromPromise,
+  fromArray,
+  buffer,
+  take,
+  mergeMap,
+  concat,
+  empty,
+  Source,
+} from 'wonka';
 import { query, write, writeOptimistic } from './operations';
 import { SchemaPredicates } from './ast';
 import { makeDict, Store, clearOptimistic } from './store';
@@ -18,6 +33,7 @@ import {
   ResolverConfig,
   OptimisticMutationConfig,
   KeyingConfig,
+  StorageAdapter,
 } from './types';
 
 type OperationResultWithMeta = OperationResult & {
@@ -87,6 +103,7 @@ export interface CacheExchangeOpts {
   optimistic?: OptimisticMutationConfig;
   keys?: KeyingConfig;
   schema?: IntrospectionQuery;
+  storage?: StorageAdapter;
 }
 
 export const cacheExchange = (opts?: CacheExchangeOpts): Exchange => ({
@@ -103,6 +120,13 @@ export const cacheExchange = (opts?: CacheExchangeOpts): Exchange => ({
     opts.keys
   );
 
+  let hydration: void | Promise<void>;
+  if (opts.storage) {
+    hydration = opts.storage.read().then(data => {
+      store.hydrateData(data, (opts as any).storage);
+    });
+  }
+
   const optimisticKeys = new Set();
   const ops: OperationMap = new Map();
   const deps: DependentOperations = makeDict();
@@ -243,16 +267,29 @@ export const cacheExchange = (opts?: CacheExchangeOpts): Exchange => ({
   };
 
   return ops$ => {
-    const sharedOps$ = pipe(
-      ops$,
+    const sharedOps$ = pipe(ops$, share);
+
+    // Buffer operations while waiting on hydration to finish
+    // If no hydration takes place we replace this stream with an empty one
+    const bufferedOps$ = hydration
+      ? pipe(
+          sharedOps$,
+          buffer(fromPromise(hydration)),
+          take(1),
+          mergeMap(fromArray)
+        )
+      : (empty as Source<Operation>);
+
+    const inputOps$ = pipe(
+      concat([bufferedOps$, sharedOps$]),
       map(addTypeNames),
       tap(optimisticUpdate),
       share
     );
 
     // Filter by operations that are cacheable and attempt to query them from the cache
     const cache$ = pipe(
-      sharedOps$,
+      inputOps$,
       filter(op => isCacheableQuery(op)),
       map(operationResultFromCache),
       share
@@ -302,7 +339,7 @@ export const cacheExchange = (opts?: CacheExchangeOpts): Exchange => ({
       forward(
         merge([
           pipe(
-            sharedOps$,
+            inputOps$,
             filter(op => !isCacheableQuery(op))
           ),
           cacheOps$,

src/store/__snapshots__/store.test.ts.snap

@@ -0,0 +1,12 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`Store with storage should be able to store and rehydrate data 1`] = `
+Object {
+  "l|Query.appointment({\\"id\\":\\"1\\"})": "Appointment:1",
+  "r|Appointment:1.__typename": "Appointment",
+  "r|Appointment:1.id": "1",
+  "r|Appointment:1.info": "urql meeting",
+  "r|Query.__typename": "Query",
+  "r|Query.appointment({\\"id\\":\\"1\\"})": undefined,
+}
+`;

src/store/data.ts

@@ -1,6 +1,12 @@
-import { Link, EntityField, FieldInfo } from '../types';
+import {
+  Link,
+  EntityField,
+  FieldInfo,
+  StorageAdapter,
+  SerializedEntries,
+} from '../types';
 import { invariant, currentDebugStack } from '../helpers/help';
-import { fieldInfoOfKey, joinKeys } from './keys';
+import { fieldInfoOfKey, joinKeys, prefixKey } from './keys';
 import { defer } from './timing';
 
 type Dict<T> = Record<string, T>;
@@ -28,6 +34,8 @@ let currentDependencies: null | Set<string> = null;
 let currentOptimisticKey: null | number = null;
 
 export const makeDict = (): any => Object.create(null);
+export const storage: { current: StorageAdapter | null } = { current: null };
+let persistanceBatch: SerializedEntries = makeDict();
 
 const makeNodeMap = <T>(): NodeMap<T> => ({
   optimistic: makeDict(),
@@ -51,9 +59,19 @@ export const initDataState = (
 /** Reset the data state after read/write is complete */
 export const clearDataState = () => {
   const data = currentData!;
+
   if (!data.gcScheduled && data.gcBatch.size > 0) {
     data.gcScheduled = true;
-    defer(() => gc(data));
+    defer(() => {
+      gc(data);
+    });
+  }
+
+  if (storage.current) {
+    defer(() => {
+      (storage.current as StorageAdapter).write(persistanceBatch);
+      persistanceBatch = makeDict();
+    });
   }
 
   currentData = null;
@@ -260,12 +278,18 @@ export const gc = (data: InMemoryData) => {
       delete data.refCount[entityKey];
       data.records.base.delete(entityKey);
       data.gcBatch.delete(entityKey);
+      if (storage.current) {
+        storage.current.remove(prefixKey('r', entityKey));
+      }
 
       // Delete all the entity's links, but also update the reference count
       // for those links (which can lead to an unrolled recursive GC of the children)
       const linkNode = data.links.base.get(entityKey);
       if (linkNode !== undefined) {
         data.links.base.delete(entityKey);
+        if (storage.current) {
+          storage.current.remove(prefixKey('l', entityKey));
+        }
         for (const key in linkNode) {
           updateRCForLink(data.gcBatch, data.refCount, linkNode[key], -1);
         }
@@ -308,10 +332,15 @@ export const readLink = (
 export const writeRecord = (
   entityKey: string,
   fieldKey: string,
-  value: EntityField
+  value: EntityField,
+  skipPersistance?: boolean
 ) => {
   updateDependencies(entityKey, fieldKey);
   setNode(currentData!.records, entityKey, fieldKey, value);
+  if (storage.current && !skipPersistance && !currentOptimisticKey) {
+    const key = prefixKey('r', joinKeys(entityKey, fieldKey));
+    persistanceBatch[key] = value;
+  }
 };
 
 export const hasField = (entityKey: string, fieldKey: string): boolean =>
@@ -322,7 +351,8 @@ export const hasField = (entityKey: string, fieldKey: string): boolean =>
 export const writeLink = (
   entityKey: string,
   fieldKey: string,
-  link: Link | undefined
+  link: Link | undefined,
+  skipPersistance?: boolean
 ) => {
   const data = currentData!;
   // Retrieve the reference counting dict or the optimistic reference locking dict
@@ -339,6 +369,10 @@ export const writeLink = (
       (data.refLock[currentOptimisticKey] = makeDict());
     links = data.links.optimistic[currentOptimisticKey];
   } else {
+    if (storage.current && !skipPersistance) {
+      const key = prefixKey('l', joinKeys(entityKey, fieldKey));
+      persistanceBatch[key] = link;
+    }
     refCount = data.refCount;
     links = data.links.base;
     gcBatch = data.gcBatch;

src/store/keys.ts

@@ -23,3 +23,6 @@ export const fieldInfoOfKey = (fieldKey: string): FieldInfo => {
 
 export const joinKeys = (parentKey: string, key: string) =>
   `${parentKey}.${key}`;
+
+/** Prefix key with its owner type Link / Record */
+export const prefixKey = (owner: 'l' | 'r', key: string) => `${owner}|${key}`;

src/store/store.test.ts

@@ -1,6 +1,6 @@
 import gql from 'graphql-tag';
 
-import { Data } from '../types';
+import { Data, StorageAdapter } from '../types';
 import { query } from '../operations/query';
 import { write, writeOptimistic } from '../operations/write';
 import * as InMemoryData from './data';
@@ -405,3 +405,56 @@ describe('Store with OptimisticMutationConfig', () => {
     });
   });
 });
+
+describe('Store with storage', () => {
+  const expectedData = {
+    __typename: 'Query',
+    appointment: {
+      __typename: 'Appointment',
+      id: '1',
+      info: 'urql meeting',
+    },
+  };
+
+  beforeEach(() => {
+    jest.useFakeTimers();
+  });
+
+  it('should be able to store and rehydrate data', () => {
+    const storage: StorageAdapter = {
+      read: jest.fn(),
+      write: jest.fn(),
+      remove: jest.fn(),
+    };
+    let store = new Store();
+
+    store.hydrateData(Object.create(null), storage);
+
+    write(
+      store,
+      {
+        query: Appointment,
+        variables: { id: '1' },
+      },
+      expectedData
+    );
+
+    expect(storage.write).not.toHaveBeenCalled();
+
+    jest.runAllTimers();
+    expect(storage.write).toHaveBeenCalled();
+
+    const serialisedStore = (storage.write as any).mock.calls[0][0];
+    expect(serialisedStore).toMatchSnapshot();
+
+    store = new Store();
+    store.hydrateData(serialisedStore, storage);
+
+    const { data } = query(store, {
+      query: Appointment,
+      variables: { id: '1' },
+    });
+
+    expect(data).toEqual(expectedData);
+  });
+});

src/store/store.ts

@@ -12,6 +12,7 @@ import {
   UpdatesConfig,
   OptimisticMutationConfig,
   KeyingConfig,
+  StorageAdapter,
 } from '../types';
 
 import { read, readFragment } from '../operations/query';
@@ -186,4 +187,22 @@ export class Store implements Cache {
   ): void {
     writeFragment(this, dataFragment, data, variables);
   }
+
+  hydrateData(data: object, adapter: StorageAdapter) {
+    InMemoryData.storage.current = adapter;
+    InMemoryData.initDataState(this.data, 0);
+    for (const key in data) {
+      const entityKey = key.slice(2, key.indexOf('.'));
+      const fieldKey = key.slice(key.indexOf('.') + 1);
+      switch (key.charCodeAt(0)) {
+        case 108:
+          InMemoryData.writeLink(entityKey, fieldKey, data[key], true);
+          break;
+        case 114:
+          InMemoryData.writeRecord(entityKey, fieldKey, data[key], true);
+          break;
+      }
+    }
+    InMemoryData.clearDataState();
+  }
 }

src/types.ts

@@ -172,6 +172,18 @@ export interface KeyingConfig {
   [typename: string]: KeyGenerator;
 }
 
+export type SerializedEntry = EntityField | Connection[] | Link;
+
+export interface SerializedEntries {
+  [key: string]: SerializedEntry;
+}
+
+export interface StorageAdapter {
+  read(): Promise<SerializedEntries>;
+  remove(key: string): Promise<void>;
+  write(data: SerializedEntries): Promise<void>;
+}
+
 export type ErrorCode =
   | 1
   | 2