fix(tanstack-react-query): use skipToken instead of overriding enabled option

.changeset/honest-streams-enable.md

@@ -0,0 +1,5 @@
+---
+'@powersync/tanstack-react-query': patch
+---
+
+Respect the user-provided `enabled` option in `useQuery` and `useQueries` instead of overriding it. Queries backed by sync streams now pause via TanStack's `skipToken` until their streams have synced, leaving `enabled` fully under the caller's control (`useSuspenseQuery` always runs, since suspense rejects `skipToken`). Also fixes a stale `streamsHaveSynced` value and a race where rows written while a query's source tables were still being resolved could be missed.

packages/tanstack-react-query/src/hooks/usePowerSyncQueries.ts

@@ -148,7 +148,7 @@ export function usePowerSyncQueries(
 
   useEffect(() => {
     const aborts = parsedQueries.map((pq, idx) => {
-      if (pq.parseError || !pq.query) {
+      if (pq.parseError || !pq.query || !tablesArr[idx]?.length) {
         return null;
       }
 
@@ -165,7 +165,8 @@ export function usePowerSyncQueries(
         },
         {
           tables: tablesArr[idx],
-          signal: abort.signal
+          signal: abort.signal,
+          triggerImmediate: true
         }
       );
 
@@ -203,5 +204,5 @@ export function usePowerSyncQueries(
       }),
       streamsHaveSynced
     };
-  }, [parsedQueries, errorsArr, tablesArr, powerSync]);
+  }, [parsedQueries, errorsArr, tablesArr, powerSync, streamsHaveSynced]);
 }

packages/tanstack-react-query/src/hooks/useQueries.ts

@@ -151,15 +151,15 @@ export function useQueries(
     return queriesInput.map((queryOptions, idx) => {
       const { query, parameters, ...rest } = queryOptions;
       const state = states[idx];
+      const queryFn = streamsHaveSynced ? (query ? state.queryFn : rest.queryFn) : Tanstack.skipToken;
 
       return {
         ...rest,
-        queryFn: query ? state.queryFn : rest.queryFn,
-        queryKey: rest.queryKey,
-        enabled: streamsHaveSynced
+        queryFn,
+        queryKey: rest.queryKey
       };
     });
-  }, [queriesInput, states]);
+  }, [queriesInput, states, streamsHaveSynced]);
 
   return Tanstack.useQueries(
     {

packages/tanstack-react-query/src/hooks/useQuery.ts

@@ -135,7 +135,7 @@ function useQueryCore<
   const { query, parameters, queryKey, streams, ...resolvedOptions } = options;
 
   const {
-    queries: [{ queryFn }],
+    queries: [{ queryFn: powerSyncQueryFn }],
     streamsHaveSynced
   } = usePowerSyncQueries(
     [
@@ -149,12 +149,15 @@ function useQueryCore<
     queryClient
   );
 
+  const isSuspense = (useQueryFn as unknown) === Tanstack.useSuspenseQuery;
+  const resolvedQueryFn = query ? powerSyncQueryFn : resolvedOptions.queryFn;
+  const queryFn = streamsHaveSynced || isSuspense ? resolvedQueryFn : Tanstack.skipToken;
+
   return useQueryFn(
     {
       ...(resolvedOptions as TQueryOptions),
       queryKey,
-      queryFn: query ? queryFn : resolvedOptions.queryFn,
-      enabled: streamsHaveSynced
+      ...((typeof queryFn === 'function' || queryFn === Tanstack.skipToken) && { queryFn })
     } as TQueryOptions,
     queryClient
   );

packages/tanstack-react-query/tests/enabled.test.tsx

@@ -0,0 +1,100 @@
+import { cleanup, renderHook, waitFor } from '@testing-library/react';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import React, { Suspense } from 'react';
+import { AbstractPowerSyncDatabase } from '@powersync/common';
+import { PowerSyncContext } from '@powersync/react';
+import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
+import { openPowerSync } from './utils';
+import { useQuery } from '../src/hooks/useQuery';
+import { useQueries } from '../src/hooks/useQueries';
+import { useSuspenseQuery } from '../src/hooks/useQuery';
+
+describe('user enabled option is respected', () => {
+  let db: AbstractPowerSyncDatabase;
+  let queryClient = new QueryClient({
+    defaultOptions: {
+      queries: {
+        retry: false
+      }
+    }
+  });
+
+  beforeEach(() => {
+    db = openPowerSync();
+    queryClient.clear();
+    vi.clearAllMocks();
+    cleanup();
+  });
+
+  const wrapper = ({ children }: { children: React.ReactNode }) => (
+    <QueryClientProvider client={queryClient}>
+      <PowerSyncContext.Provider value={db}>{children}</PowerSyncContext.Provider>
+    </QueryClientProvider>
+  );
+
+  it('useQuery: does not run when user passes enabled: false (no streams)', async () => {
+    const getAllSpy = vi.spyOn(db, 'getAll');
+
+    const { result } = renderHook(
+      () =>
+        useQuery({
+          queryKey: ['lists'],
+          query: 'SELECT * FROM lists',
+          enabled: false
+        }),
+      { wrapper }
+    );
+
+    await new Promise((resolve) => setTimeout(resolve, 300));
+
+    expect(result.current.status).toBe('pending');
+    expect(result.current.fetchStatus).toBe('idle');
+    expect(result.current.data).toBeUndefined();
+    expect(getAllSpy).not.toHaveBeenCalledWith('SELECT * FROM lists', expect.anything());
+  });
+
+  it('useQueries: does not run an entry where user passes enabled: false (no streams)', async () => {
+    const getAllSpy = vi.spyOn(db, 'getAll');
+
+    const { result } = renderHook(
+      () =>
+        useQueries({
+          queries: [{ queryKey: ['lists'], query: 'SELECT * FROM lists', enabled: false }]
+        }),
+      { wrapper }
+    );
+
+    await new Promise((resolve) => setTimeout(resolve, 300));
+
+    expect(result.current[0].status).toBe('pending');
+    expect(result.current[0].fetchStatus).toBe('idle');
+    expect(result.current[0].data).toBeUndefined();
+    expect(getAllSpy).not.toHaveBeenCalledWith('SELECT * FROM lists', expect.anything());
+  });
+
+  it('useSuspenseQuery: with an unsynced waitForStream stream does not error with skipToken and still resolves', async () => {
+    const consoleErrorSpy = vi.spyOn(console, 'error');
+
+    const suspenseWrapper = ({ children }: { children: React.ReactNode }) => (
+      <QueryClientProvider client={queryClient}>
+        <PowerSyncContext.Provider value={db}>
+          <Suspense fallback={<div data-testid="fallback">loading</div>}>{children}</Suspense>
+        </PowerSyncContext.Provider>
+      </QueryClientProvider>
+    );
+
+    const { result } = renderHook(
+      () =>
+        useSuspenseQuery({
+          queryKey: ['suspense-lists'],
+          query: 'SELECT * FROM lists',
+          streams: [{ name: 'a', waitForStream: true }]
+        }),
+      { wrapper: suspenseWrapper }
+    );
+
+    await waitFor(() => expect(result.current.data).toBeDefined(), { timeout: 1000, interval: 100 });
+
+    expect(consoleErrorSpy).not.toHaveBeenCalledWith('skipToken is not allowed for useSuspenseQuery');
+  });
+});

packages/tanstack-react-query/tests/usePowerSyncQueries.test.tsx

@@ -0,0 +1,141 @@
+import { cleanup, renderHook, waitFor } from '@testing-library/react';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import React from 'react';
+import { AbstractPowerSyncDatabase, SyncStatus } from '@powersync/common';
+import { PowerSyncContext } from '@powersync/react';
+import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
+import { openPowerSync } from './utils';
+import * as Tanstack from '@tanstack/react-query';
+import { usePowerSyncQueries } from '../src/hooks/usePowerSyncQueries';
+
+describe('usePowerSyncQueries bug fixes', () => {
+  let db: AbstractPowerSyncDatabase;
+  let queryClient = new QueryClient({
+    defaultOptions: {
+      queries: {
+        retry: false
+      }
+    }
+  });
+
+  beforeEach(() => {
+    db = openPowerSync();
+    queryClient.clear();
+    vi.clearAllMocks();
+    cleanup();
+  });
+
+  const wrapper = ({ children }: { children: React.ReactNode }) => (
+    <QueryClientProvider client={queryClient}>
+      <PowerSyncContext.Provider value={db}>{children}</PowerSyncContext.Provider>
+    </QueryClientProvider>
+  );
+
+  const syncedStatus = () =>
+    new SyncStatus({
+      dataFlow: {
+        internalStreamSubscriptions: [
+          {
+            name: 'a',
+            parameters: null,
+            progress: { total: 0, downloaded: 0 },
+            active: true,
+            is_default: false,
+            has_explicit_subscription: true,
+            expires_at: null,
+            last_synced_at: 1234,
+            priority: 1
+          }
+        ]
+      }
+    });
+
+  describe('usePowerSyncQueries ', () => {
+    it('updated returned streamsHaveSynced once a waitForStream stream syncs', async () => {
+      const stableQueries = [
+        {
+          query: 'SELECT name FROM lists',
+          queryKey: ['bug1'],
+          streams: [{ name: 'a', waitForStream: true }]
+        }
+      ];
+
+      const { result, unmount } = renderHook(() => usePowerSyncQueries(stableQueries, queryClient), { wrapper });
+
+      await waitFor(() => expect(result.current.streamsHaveSynced).toBe(false), { timeout: 1000, interval: 50 });
+
+      db.currentStatus = syncedStatus();
+      db.iterateListeners((l) => l.statusChanged?.(syncedStatus()));
+
+      await waitFor(() => expect(result.current.streamsHaveSynced).toBe(true), { timeout: 1000, interval: 50 });
+
+      await waitFor(() => expect(result.current.queries[0].tables).toContain('lists'), {
+        timeout: 1000,
+        interval: 50
+      });
+
+      unmount();
+    });
+
+    it('picks up rows written before the source tables finished resolving', async () => {
+      let releaseResolve!: () => void;
+      const resolveGate = new Promise<void>((resolve) => {
+        releaseResolve = resolve;
+      });
+
+      const realResolveTables = db.resolveTables.bind(db);
+      const resolveSpy = vi.spyOn(db, 'resolveTables').mockImplementation(async (sql, params) => {
+        await resolveGate;
+        return realResolveTables(sql, params);
+      });
+
+      const realRegisterListener = db.registerListener.bind(db);
+      const registerListenerSpy = vi
+        .spyOn(db, 'registerListener')
+        .mockImplementation((listener: Parameters<typeof realRegisterListener>[0]) => {
+          if (listener && (listener as { schemaChanged?: unknown }).schemaChanged) {
+            const { schemaChanged: _omit, ...rest } = listener as Record<string, unknown>;
+            return realRegisterListener(rest as Parameters<typeof realRegisterListener>[0]);
+          }
+          return realRegisterListener(listener);
+        });
+
+      const stableQueries = [{ query: 'SELECT name FROM lists ORDER BY name', queryKey: ['bug2'] }];
+
+      const { result, unmount } = renderHook(
+        () => {
+          const { queries } = usePowerSyncQueries(stableQueries, queryClient);
+          return Tanstack.useQuery(
+            {
+              queryKey: ['bug2'],
+              queryFn: queries[0].queryFn as () => Promise<{ name: string }[]>,
+              staleTime: Infinity,
+              refetchOnMount: false,
+              refetchOnWindowFocus: false,
+              refetchOnReconnect: false
+            },
+            queryClient
+          );
+        },
+        { wrapper }
+      );
+
+      await waitFor(() => expect(result.current.data).toEqual([]), { timeout: 2000, interval: 50 });
+
+      await db.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['from-first-sync']);
+
+      await new Promise((r) => setTimeout(r, 150));
+
+      releaseResolve();
+
+      await waitFor(() => expect(result.current.data).toEqual([{ name: 'from-first-sync' }]), {
+        timeout: 2000,
+        interval: 50
+      });
+
+      resolveSpy.mockRestore();
+      registerListenerSpy.mockRestore();
+      unmount();
+    });
+  });
+});

packages/tanstack-react-query/tests/utils.ts

@@ -1,5 +1,6 @@
 import * as commonSdk from '@powersync/common';
 
+import { cleanup } from '@testing-library/react';
 import { PowerSyncDatabase } from '@powersync/web';
 import { onTestFinished } from 'vitest';
 
@@ -14,6 +15,8 @@ export const openPowerSync = () => {
   });
 
   onTestFinished(async () => {
+    cleanup();
+    await new Promise((resolve) => setTimeout(resolve, 100));
     await db.disconnectAndClear();
     await db.close();
   });