Add support for live queries

Using the @LiVe query directive, we can ask Gadget to give us back a stream of data as it changes from the backend! We can use the graphql-ws transport we already have setup and productionized for this, and we can use urql's ability to rerender at will to trigger a nice rerender. To start, live queries are only available in the React hooks for rerendering.

Author: airhorns

packages/api-client-core/package.json

@@ -28,6 +28,9 @@
     "prerelease": "gitpkg publish"
   },
   "dependencies": {
+    "@n1ru4l/graphql-live-query": "^0.10.0",
+    "@n1ru4l/graphql-live-query-patch-jsondiffpatch": "^0.8.0",
+    "@n1ru4l/push-pull-async-iterable-iterator": "^3.2.0",
     "@urql/core": "^4.0.10",
     "cross-fetch": "^3.1.5",
     "graphql": "^16.8.1",

packages/api-client-core/spec/mockUrqlClient.ts

@@ -1,6 +1,7 @@
 import type { Client, GraphQLRequest, OperationContext, OperationResult, OperationResultSource } from "@urql/core";
 import { createRequest, makeErrorResult } from "@urql/core";
-import type { DocumentNode, OperationDefinitionNode } from "graphql";
+import type { DocumentNode, ExecutionResult, OperationDefinitionNode } from "graphql";
+import type { SubscribePayload, Client as SubscriptionClient, Sink as SubscriptionSink } from "graphql-ws";
 import { find, findLast } from "lodash";
 import { act } from "react-dom/test-utils";
 import type { Sink, Source, Subject } from "wonka";
@@ -168,7 +169,50 @@ export const createMockUrqlClient = (assertions?: {
   } as MockUrqlClient;
 };
 
+export interface MockSubscription {
+  payload: SubscribePayload;
+  sink: SubscriptionSink<ExecutionResult<any, any>>;
+  push: (result: ExecutionResult<any, any>) => void;
+  disposed: boolean;
+}
+export type MockSubscribeFn = ((payload: SubscribePayload, sink: SubscriptionSink<ExecutionResult<any, any>>) => () => void) & {
+  subscriptions: MockSubscription[];
+};
+
+export interface MockGraphQLWSClient extends SubscriptionClient {
+  subscribe: MockSubscribeFn;
+}
+
+/**
+ * Create a new function for mocking subscriptions passed to graphql-ws
+ */
+function newMockSubscribeFn(): MockSubscribeFn {
+  const subscriptions: MockSubscription[] = [];
+
+  const fn: SubscriptionClient["subscribe"] = (payload: SubscribePayload, sink: SubscriptionSink<ExecutionResult<any, any>>) => {
+    const subscription: MockSubscription = {
+      payload,
+      sink,
+      disposed: false,
+      push: (result) => {
+        act(() => {
+          sink.next(result);
+        });
+      },
+    };
+
+    subscriptions.push(subscription);
+
+    return () => {
+      subscription.disposed = true;
+    };
+  };
+
+  return Object.assign(fn, { subscriptions });
+}
+
 export const mockUrqlClient = createMockUrqlClient();
+export const mockGraphQLWSClient = {} as MockGraphQLWSClient;
 
 beforeEach(() => {
   const fetch = newMockFetchFn();
@@ -180,6 +224,8 @@ beforeEach(() => {
     fetch,
   };
   mockUrqlClient.mockFetch = fetch;
+
+  mockGraphQLWSClient.subscribe = newMockSubscribeFn();
 });
 
 afterEach(() => {

packages/api-client-core/spec/operationBuilders.spec.ts

@@ -42,6 +42,27 @@ describe("operation builders", () => {
         }
       `);
     });
+
+    test("findOneOperation should build a live query for a model", () => {
+      expect(findOneOperation("widget", "123", { __typename: true, id: true, state: true }, "widget", { live: true }))
+        .toMatchInlineSnapshot(`
+        {
+          "query": "query widget($id: GadgetID!) @live {
+          widget(id: $id) {
+            __typename
+            id
+            state
+          }
+          gadgetMeta {
+            hydrations(modelName: "widget")
+          }
+        }",
+          "variables": {
+            "id": "123",
+          },
+        }
+      `);
+    });
   });
 
   describe("findManyOperation", () => {
@@ -209,6 +230,35 @@ describe("operation builders", () => {
         }
       `);
     });
+
+    test("findManyOperation should build a live findMany query for a model", () => {
+      expect(findManyOperation("widgets", { __typename: true, id: true, state: true }, "widget", { live: true })).toMatchInlineSnapshot(`
+        {
+          "query": "query widgets($after: String, $first: Int, $before: String, $last: Int) @live {
+          widgets(after: $after, first: $first, before: $before, last: $last) {
+            pageInfo {
+              hasNextPage
+              hasPreviousPage
+              startCursor
+              endCursor
+            }
+            edges {
+              cursor
+              node {
+                __typename
+                id
+                state
+              }
+            }
+          }
+          gadgetMeta {
+            hydrations(modelName: "widget")
+          }
+        }",
+          "variables": {},
+        }
+      `);
+    });
   });
 
   describe("findOneByFieldOperation", () => {
@@ -287,6 +337,43 @@ describe("operation builders", () => {
         }
       `);
     });
+
+    test("findOneByFieldOperation should build a live query", () => {
+      expect(findOneByFieldOperation("widget", "foo", "bar", { __typename: true, id: true, state: true }, "widget", { live: true }))
+        .toMatchInlineSnapshot(`
+        {
+          "query": "query widget($after: String, $first: Int, $before: String, $last: Int, $filter: [WidgetFilter!]) @live {
+          widget(after: $after, first: $first, before: $before, last: $last, filter: $filter) {
+            pageInfo {
+              hasNextPage
+              hasPreviousPage
+              startCursor
+              endCursor
+            }
+            edges {
+              cursor
+              node {
+                __typename
+                id
+                state
+              }
+            }
+          }
+          gadgetMeta {
+            hydrations(modelName: "widget")
+          }
+        }",
+          "variables": {
+            "filter": {
+              "foo": {
+                "equals": "bar",
+              },
+            },
+            "first": 2,
+          },
+        }
+      `);
+    });
   });
 
   describe("actionOperation", () => {

packages/api-client-core/src/GadgetConnection.ts

@@ -1,7 +1,9 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
+import { isLiveQueryOperationDefinitionNode } from "@n1ru4l/graphql-live-query";
+import { applyLiveQueryJSONDiffPatch } from "@n1ru4l/graphql-live-query-patch-jsondiffpatch";
+import { applyAsyncIterableIteratorToSink, makeAsyncIterableIteratorFromSink } from "@n1ru4l/push-pull-async-iterable-iterator";
 import type { ClientOptions, RequestPolicy } from "@urql/core";
 import { Client, cacheExchange, fetchExchange, subscriptionExchange } from "@urql/core";
-
 import type { ExecutionResult } from "graphql";
 import type { Sink, Client as SubscriptionClient, ClientOptions as SubscriptionClientOptions } from "graphql-ws";
 import { CloseCode, createClient as createSubscriptionClient } from "graphql-ws";
@@ -80,7 +82,9 @@ export class GadgetConnection {
 
   // the base client using HTTP requests that non-transactional operations will use
   private baseClient: Client;
-  private baseSubscriptionClient?: SubscriptionClient;
+
+  /** @private (but accessible for testing purposes) */
+  baseSubscriptionClient?: SubscriptionClient;
 
   // the transactional websocket client that will be used inside a transaction block
   private currentTransaction: GadgetTransaction | null = null;
@@ -341,10 +345,9 @@ export class GadgetConnection {
     if (typeof window != "undefined") {
       exchanges.push(cacheExchange);
     }
-
     exchanges.push(
       ...this.exchanges.beforeAsync,
-      fetchExchange,
+      // standard subscriptions for normal GraphQL subscriptions
       subscriptionExchange({
         forwardSubscription: (request) => {
           return {
@@ -359,6 +362,31 @@ export class GadgetConnection {
           };
         },
       }),
+      // another subscription exchange for live queries
+      // live queries pass through the same WS client, but use jsondiffs for patching in results
+      subscriptionExchange({
+        isSubscriptionOperation: (request) => {
+          return request.query.definitions.some((definition) => isLiveQueryOperationDefinitionNode(definition, request.variables as any));
+        },
+        forwardSubscription: (request) => {
+          return {
+            subscribe: (sink) => {
+              const input = { ...request, query: request.query || "" };
+              return {
+                unsubscribe: applyAsyncIterableIteratorToSink(
+                  applyLiveQueryJSONDiffPatch(
+                    makeAsyncIterableIteratorFromSink<ExecutionResult>((sink) =>
+                      this.getBaseSubscriptionClient().subscribe(input, sink as Sink<ExecutionResult>)
+                    )
+                  ),
+                  sink
+                ),
+              };
+            },
+          };
+        },
+      }),
+      fetchExchange,
       ...this.exchanges.afterAll
     );
 

packages/api-client-core/src/operationBuilders.ts

@@ -2,7 +2,7 @@ import type { FieldSelection as BuilderFieldSelection, BuilderOperation, Variabl
 import { Call, Var, compileWithVariableValues } from "tiny-graphql-query-compiler";
 import type { FieldSelection } from "./FieldSelection.js";
 import { filterTypeName, sortTypeName } from "./support.js";
-import type { FindManyOptions, SelectionOptions, VariablesOptions } from "./types.js";
+import type { BaseFindOptions, FindManyOptions, VariablesOptions } from "./types.js";
 
 const hydrationOptions = (modelApiIdentifier: string): BuilderFieldSelection => {
   return {
@@ -23,12 +23,17 @@ const fieldSelectionToQueryCompilerFields = (selection: FieldSelection, includeT
 
 export type FindFirstPaginationOptions = Omit<FindManyOptions, "first" | "last" | "before" | "after">;
 
+const directivesForOptions = (options?: BaseFindOptions | null) => {
+  if (options?.live) return ["@live"];
+  return undefined;
+};
+
 export const findOneOperation = (
   operation: string,
   id: string | undefined,
   defaultSelection: FieldSelection,
   modelApiIdentifier: string,
-  options?: SelectionOptions | null
+  options?: BaseFindOptions | null
 ) => {
   const variables: Record<string, Variable> = {};
   if (typeof id !== "undefined") variables.id = Var({ type: "GadgetID!", value: id });
@@ -39,6 +44,7 @@ export const findOneOperation = (
       [operation]: Call(variables, fieldSelectionToQueryCompilerFields(options?.select || defaultSelection, true)),
       ...hydrationOptions(modelApiIdentifier),
     },
+    directives: directivesForOptions(options),
   });
 };
 
@@ -48,10 +54,10 @@ export const findOneByFieldOperation = (
   fieldValue: string,
   defaultSelection: FieldSelection,
   modelApiIdentifier: string,
-  options?: SelectionOptions | null
+  options?: BaseFindOptions | null
 ) => {
   return findManyOperation(operation, defaultSelection, modelApiIdentifier, {
-    select: options?.select,
+    ...options,
     first: 2,
     filter: {
       [fieldName]: {
@@ -91,6 +97,7 @@ export const findManyOperation = (
       ),
       ...hydrationOptions(modelApiIdentifier),
     },
+    directives: directivesForOptions(options),
   });
 };
 
@@ -115,7 +122,7 @@ export const actionOperation = (
   modelApiIdentifier: string,
   modelSelectionField: string,
   variables: VariablesOptions,
-  options?: SelectionOptions | null,
+  options?: BaseFindOptions | null,
   namespace?: string | null,
   isBulkAction?: boolean | null,
   hasReturnType?: boolean | null
@@ -144,12 +151,18 @@ export const actionOperation = (
       ...fields,
       ...hydrationOptions(modelApiIdentifier),
     },
+    directives: directivesForOptions(options),
   };
 
   return compileWithVariableValues(actionOperation);
 };
 
-export const globalActionOperation = (operation: string, variables: VariablesOptions, namespace?: string | null) => {
+export const globalActionOperation = (
+  operation: string,
+  variables: VariablesOptions,
+  namespace?: string | null,
+  options?: { live?: boolean }
+) => {
   let fields: BuilderFieldSelection = {
     [operation]: Call(variableOptionsToVariables(variables), {
       success: true,
@@ -170,5 +183,6 @@ export const globalActionOperation = (operation: string, variables: VariablesOpt
     type: "mutation",
     name: operation,
     fields,
+    directives: directivesForOptions(options),
   });
 };