Implement liveQueryExchange to prevent duplicate live query execution

- Add liveQueryExchange that tracks operations by key + variables hash
- Prevents duplicate execution of live queries with identical parameters
- Allows re-execution when variables change or after teardown
- Add comprehensive tests for exchange behavior
- Update GadgetConnection to use the new exchange
- Add React test cases for live query re-execution scenarios

Addresses feedback from PR #864 by moving exchange to separate file
with proper documentation and comprehensive test coverage.

Co-Authored-By: Harry Brundage <[email protected]>

Author: devin-ai-integration[bot]

packages/api-client-core/spec/exchanges/liveQueryExchange.spec.ts

@@ -0,0 +1,178 @@
+import { jest } from "@jest/globals";
+import type { Operation } from "@urql/core";
+import { gql, makeOperation } from "@urql/core";
+import { fromArray, pipe, toArray } from "wonka";
+import { liveQueryExchange } from "../../src/exchanges/liveQueryExchange.js";
+
+describe("liveQueryExchange", () => {
+  const mockForward = jest.fn((ops$: any) => ops$);
+  let exchange: any;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    exchange = liveQueryExchange({ forward: mockForward, client: {} as any, dispatchDebug: jest.fn() });
+  });
+
+  const createLiveQuery = (key = 1, variables: any = {}): Operation => {
+    return makeOperation(
+      "query",
+      {
+        key,
+        query: gql`
+          query TestQuery @live {
+            users {
+              id
+              name
+            }
+          }
+        `,
+        variables,
+      },
+      {} as any
+    );
+  };
+
+  const createRegularQuery = (key = 2): Operation => {
+    return makeOperation(
+      "query",
+      {
+        key,
+        query: gql`
+          query TestQuery {
+            users {
+              id
+              name
+            }
+          }
+        `,
+        variables: {},
+      },
+      {} as any
+    );
+  };
+
+  const createTeardown = (key = 1, variables: any = {}): Operation => {
+    return makeOperation(
+      "teardown",
+      {
+        key,
+        query: gql`
+          query TestQuery @live {
+            users {
+              id
+              name
+            }
+          }
+        `,
+        variables,
+      },
+      {} as any
+    );
+  };
+
+  test("allows first live query to pass through", () => {
+    const liveQuery = createLiveQuery(1);
+    const operations$ = fromArray([liveQuery]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(1);
+    expect(results[0]).toBe(liveQuery);
+  });
+
+  test("blocks duplicate live query execution with same variables", () => {
+    const liveQuery1 = createLiveQuery(1, { filter: { id: "1" } });
+    const liveQuery2 = createLiveQuery(1, { filter: { id: "1" } }); // Same key and variables
+    const operations$ = fromArray([liveQuery1, liveQuery2]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(1);
+    expect(results[0]).toBe(liveQuery1);
+  });
+
+  test("allows regular queries to pass through unchanged", () => {
+    const regularQuery1 = createRegularQuery(1);
+    const regularQuery2 = createRegularQuery(1); // Same key but not live
+    const operations$ = fromArray([regularQuery1, regularQuery2]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(2);
+    expect(results[0]).toBe(regularQuery1);
+    expect(results[1]).toBe(regularQuery2);
+  });
+
+  test("allows teardown operations to pass through", () => {
+    const liveQuery = createLiveQuery(1);
+    const teardown = createTeardown(1);
+    const operations$ = fromArray([liveQuery, teardown]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(2);
+    expect(results[0]).toBe(liveQuery);
+    expect(results[1]).toBe(teardown);
+  });
+
+  test("allows live query re-execution after teardown", () => {
+    const variables = { filter: { id: "1" } };
+    const liveQuery1 = createLiveQuery(1, variables);
+    const teardown = createTeardown(1, variables); // Same variables as the query
+    const liveQuery2 = createLiveQuery(1, variables); // Same key and variables after teardown
+    const operations$ = fromArray([liveQuery1, teardown, liveQuery2]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(3);
+    expect(results[0]).toBe(liveQuery1);
+    expect(results[1]).toBe(teardown);
+    expect(results[2]).toBe(liveQuery2);
+  });
+
+  test("handles multiple live queries with different keys", () => {
+    const liveQuery1 = createLiveQuery(1, { filter: { id: "1" } });
+    const liveQuery2 = createLiveQuery(2, { filter: { id: "2" } });
+    const operations$ = fromArray([liveQuery1, liveQuery2]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(2);
+    expect(results[0]).toBe(liveQuery1);
+    expect(results[1]).toBe(liveQuery2);
+  });
+
+  test("allows non-query operations for live queries to pass through", () => {
+    const liveQuery = createLiveQuery(1);
+    const mutation = makeOperation(
+      "mutation",
+      {
+        key: 1,
+        query: gql`
+          mutation TestMutation @live {
+            updateUser(id: "1", user: { name: "test" }) {
+              id
+              name
+            }
+          }
+        `,
+        variables: {},
+      },
+      {} as any
+    );
+    const operations$ = fromArray([liveQuery, mutation]);
+
+    const result$ = exchange(operations$);
+    const results = pipe(result$, toArray);
+
+    expect(results).toHaveLength(2);
+    expect(results[0]).toBe(liveQuery);
+    expect(results[1]).toBe(mutation);
+  });
+});

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

@@ -12,6 +12,7 @@ import { BrowserSessionStorageType } from "./ClientOptions.js";
 import { GadgetTransaction, TransactionRolledBack } from "./GadgetTransaction.js";
 import type { BrowserStorage } from "./InMemoryStorage.js";
 import { InMemoryStorage } from "./InMemoryStorage.js";
+import { liveQueryExchange } from "./exchanges/liveQueryExchange.js";
 import { operationNameExchange } from "./exchanges/operationNameExchange.js";
 import { addUrlParams, urlParamExchange } from "./exchanges/urlParamExchange.js";
 import {
@@ -368,6 +369,7 @@ export class GadgetConnection {
     // apply urql's default caching behaviour when client side (but skip it server side)
     if (typeof window != "undefined") {
       exchanges.push(cacheExchange);
+      exchanges.push(liveQueryExchange);
     }
     exchanges.push(
       ...this.exchanges.beforeAsync,
@@ -442,17 +444,6 @@ export class GadgetConnection {
     });
     (client as any)[$gadgetConnection] = this;
 
-    const reexecuteOperation = client.reexecuteOperation.bind(client);
-    client.reexecuteOperation = (operation) => {
-      if (operation.query.definitions.some(isLiveQueryOperationDefinitionNode)) {
-        // live queries don't cleanup properly when reexecuted, plus
-        // they should never need to be reexecuted since they receive
-        // updates from the server, so we noop here
-        return;
-      }
-      reexecuteOperation(operation);
-    };
-
     return client;
   }
 
@@ -665,6 +656,6 @@ const getLiveDirectiveNode = (input: DefinitionNode): Maybe<DirectiveNode> => {
   return input.directives?.find((d) => d.name.value === "live");
 };
 
-const isLiveQueryOperationDefinitionNode = (input: DefinitionNode): input is OperationDefinitionNode => {
+export const isLiveQueryOperationDefinitionNode = (input: DefinitionNode): input is OperationDefinitionNode => {
   return !!getLiveDirectiveNode(input);
 };

packages/api-client-core/src/exchanges/liveQueryExchange.ts

@@ -0,0 +1,42 @@
+import type { Exchange, Operation } from "@urql/core";
+import { filter, merge, pipe, tap } from "wonka";
+import { isLiveQueryOperationDefinitionNode } from "../GadgetConnection.js";
+
+/**
+ * Exchange that prevents duplicate execution of live queries while allowing proper teardown and re-establishment.
+ *
+ * Live queries are long-running special subscriptions that receive real-time updates from the server. When mutations occur, urql's cache exchange tries to re-execute any mounted queries that could be affected and so re-executes live queries. Live queries automatically update their own data using their own persistent connection, so they shouldn't be re-executed if already mounted.
+ */
+export const liveQueryExchange: Exchange = ({ forward }) => {
+  const executed = new Set<number>();
+
+  const getOperationId = (op: Operation<any, any>) => {
+    return op.key;
+  };
+
+  return (operations$) => {
+    const notLive = pipe(
+      operations$,
+      filter((op) => !op.query.definitions.some(isLiveQueryOperationDefinitionNode))
+    );
+
+    const live = pipe(
+      operations$,
+      filter((op) => op.query.definitions.some(isLiveQueryOperationDefinitionNode)),
+      filter((op) => {
+        const opId = getOperationId(op);
+        return !executed.has(opId) || op.kind !== "query";
+      }),
+      tap((op) => {
+        const opId = getOperationId(op);
+        if (op.kind === "query") {
+          executed.add(opId);
+        } else if (op.kind === "teardown") {
+          executed.delete(opId);
+        }
+      })
+    );
+
+    return forward(merge([live, notLive]));
+  };
+};

packages/react/.changeset/clever-clouds-march.md

@@ -0,0 +1,8 @@
+---
+"@gadgetinc/api-client-core": patch
+"@gadgetinc/react": patch
+---
+
+Ensure live queries correctly re-execute when input variables change
+
+Previously, mounted live queries in React would erroneously not re-connect to the server when their input variables changed, like the filter, sort, or selection. The websocket would confusingly stay open, but the client side code listening for changes would stop listening and not restart. This has now been fixed and live queries will correctly re-establish a new websocket connection with new variables when the variables change.