(feat): Tanstack Query Sync Streams (#866)

Author: Chriztiaan

.changeset/nervous-taxis-explode.md

@@ -0,0 +1,5 @@
+---
+'@powersync/react': minor
+---
+
+Exporting `useAllSyncStreamsHaveSynced()` member.

.changeset/tasty-meals-tap.md

@@ -0,0 +1,5 @@
+---
+'@powersync/tanstack-react-query': minor
+---
+
+Added sync stream support to the `useQuery()` and `useQueries()` hooks.

packages/react/src/hooks/streams.ts

@@ -66,7 +66,7 @@ export function useSyncStream(options: UseSyncStreamOptions): SyncStreamStatus |
 }
 
 /**
- * @internal
+ * Returns `true` once all streams in the array have synced at least once.
  */
 export function useAllSyncStreamsHaveSynced(
   db: AbstractPowerSyncDatabase,

packages/react/src/index.ts

@@ -5,8 +5,8 @@ export * from './hooks/PowerSyncContext.js';
 export { SuspenseQueryResult } from './hooks/suspense/SuspenseQueryResult.js';
 export { useSuspenseQuery } from './hooks/suspense/useSuspenseQuery.js';
 export { useWatchedQuerySuspenseSubscription } from './hooks/suspense/useWatchedQuerySuspenseSubscription.js';
-export { useSyncStream, UseSyncStreamOptions } from './hooks/streams.js';
+export { useAllSyncStreamsHaveSynced, useSyncStream, UseSyncStreamOptions } from './hooks/streams.js';
 export { useStatus } from './hooks/useStatus.js';
 export { useQuery } from './hooks/watched/useQuery.js';
 export { useWatchedQuerySubscription } from './hooks/watched/useWatchedQuerySubscription.js';
-export { AdditionalOptions } from './hooks/watched/watch-types.js';
+export { AdditionalOptions, QuerySyncStreamOptions } from './hooks/watched/watch-types.js';

packages/tanstack-react-query/package.json

@@ -41,6 +41,7 @@
     "@tanstack/react-query": "^5.70.0"
   },
   "devDependencies": {
+    "@powersync/web": "workspace:*",
     "@testing-library/react": "^15.0.2",
     "@types/react": "18.3.1",
     "jsdom": "catalog:",

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

@@ -1,21 +1,25 @@
 import { type CompilableQuery, parseQuery } from '@powersync/common';
-import { usePowerSync } from '@powersync/react';
+import { QuerySyncStreamOptions, useAllSyncStreamsHaveSynced, usePowerSync } from '@powersync/react';
 import { useEffect, useState, useCallback, useMemo } from 'react';
 import * as Tanstack from '@tanstack/react-query';
 
 export type UsePowerSyncQueriesInput = {
   query?: string | CompilableQuery<unknown>;
   parameters?: unknown[];
   queryKey: Tanstack.QueryKey;
+  streams?: QuerySyncStreamOptions[];
 }[];
 
 export type UsePowerSyncQueriesOutput = {
-  sqlStatement: string;
-  queryParameters: unknown[];
-  tables: string[];
-  error?: Error;
-  queryFn: () => Promise<unknown[]>;
-}[];
+  queries: {
+    sqlStatement: string;
+    queryParameters: unknown[];
+    tables: string[];
+    error?: Error;
+    queryFn: () => Promise<unknown[]>;
+  }[];
+  streamsHaveSynced: boolean;
+};
 
 export function usePowerSyncQueries(
   queries: UsePowerSyncQueriesInput,
@@ -35,6 +39,14 @@ export function usePowerSyncQueries(
     });
   }, []);
 
+  // Collect all streams from all query entries into a single flat list for sync tracking.
+  const allStreams = useMemo(() => {
+    const streams = queries.flatMap(q => q.streams ?? []);
+    return streams.length > 0 ? streams : undefined;
+  }, [queries]);
+
+  const streamsHaveSynced = useAllSyncStreamsHaveSynced(powerSync, allStreams);
+
   const updateErrorsArr = useCallback((error: Error | undefined, idx: number) => {
     setErrorsArr((prev) => {
       if (prev[idx]?.message === error?.message) return prev;
@@ -164,29 +176,33 @@ export function usePowerSyncQueries(
   }, [powerSync, queryClient, tablesArr, updateErrorsArr, stringifiedQueryKeys]);
 
   return useMemo(() => {
-    return parsedQueries.map((pq, idx) => {
-      const error = errorsArr[idx] || pq.parseError;
-
-      const queryFn = async () => {
-        if (error) throw error;
-        if (!pq.query) throw new Error('No query provided');
 
-        try {
-          return typeof pq.query === 'string'
-            ? await powerSync.getAll(pq.sqlStatement, pq.queryParameters)
-            : await pq.query.execute();
-        } catch (e) {
-          throw e;
-        }
-      };
-
-      return {
-        sqlStatement: pq.sqlStatement,
-        queryParameters: pq.queryParameters,
-        tables: tablesArr[idx],
-        error,
-        queryFn
-      };
-    });
+    return {
+      queries: parsedQueries.map((pq, idx) => {
+        const error = errorsArr[idx] || pq.parseError;
+  
+        const queryFn = async () => {
+          if (error) throw error;
+          if (!pq.query) throw new Error('No query provided');
+  
+          try {
+            return typeof pq.query === 'string'
+              ? await powerSync.getAll(pq.sqlStatement, pq.queryParameters)
+              : await pq.query.execute();
+          } catch (e) {
+            throw e;
+          }
+        };
+  
+        return {
+          sqlStatement: pq.sqlStatement,
+          queryParameters: pq.queryParameters,
+          tables: tablesArr[idx],
+          error,
+          queryFn
+        };
+      }),
+      streamsHaveSynced
+    };
   }, [parsedQueries, errorsArr, tablesArr, powerSync]);
 }

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

@@ -1,12 +1,24 @@
 import { type CompilableQuery } from '@powersync/common';
-import { usePowerSync } from '@powersync/react';
+import { QuerySyncStreamOptions, usePowerSync } from '@powersync/react';
 import * as Tanstack from '@tanstack/react-query';
 import { useMemo } from 'react';
 import { usePowerSyncQueries } from './usePowerSyncQueries.js';
 
 export type PowerSyncQueryOptions<T> = {
   query?: string | CompilableQuery<T>;
   parameters?: any[];
+  /**
+   * An optional array of sync streams (with names and parameters) backing the query.
+   *
+   * When set, `useQuery` will subscribe to those streams (and automatically handle unsubscribing from them, too).
+   *
+   * If {@link QuerySyncStreamOptions} is set on a stream, `useQuery` will remain in a loading state until that stream
+   * is marked as {@link SyncSubscriptionDescription.hasSynced}. This ensures the query is not missing rows that haven't
+   * been downloaded.
+   * Note however that after an initial sync, the query will not block itself while new rows are downloading. Instead,
+   * consistent sync snapshots will be made available as they've been processed by PowerSync.
+   **/
+  streams?: QuerySyncStreamOptions[];
 };
 
 export type PowerSyncQueryOption<T = unknown[]> = Tanstack.UseQueryOptions<T[]> & PowerSyncQueryOptions<T>;
@@ -126,12 +138,13 @@ export function useQueries(
       queriesInput.map((queryOptions) => ({
         query: queryOptions.query,
         parameters: queryOptions.parameters,
-        queryKey: queryOptions.queryKey
+        queryKey: queryOptions.queryKey,
+        streams: queryOptions.streams
       })),
     [queriesInput]
   );
 
-  const states = usePowerSyncQueries(powerSyncQueriesInput, queryClient);
+  const {queries: states, streamsHaveSynced} = usePowerSyncQueries(powerSyncQueriesInput, queryClient);
 
   const queries = useMemo(() => {
     return queriesInput.map((queryOptions, idx) => {
@@ -141,7 +154,8 @@ export function useQueries(
       return {
         ...rest,
         queryFn: query ? state.queryFn : rest.queryFn,
-        queryKey: rest.queryKey
+        queryKey: rest.queryKey,
+        enabled: streamsHaveSynced
       };
     });
   }, [queriesInput, states]);

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

@@ -1,11 +1,23 @@
 import { type CompilableQuery } from '@powersync/common';
-import { usePowerSync } from '@powersync/react';
+import { QuerySyncStreamOptions, usePowerSync } from '@powersync/react';
 import * as Tanstack from '@tanstack/react-query';
 import { usePowerSyncQueries } from './usePowerSyncQueries.js';
 
 export type PowerSyncQueryOptions<T> = {
   query?: string | CompilableQuery<T>;
   parameters?: any[];
+  /**
+   * An optional array of sync streams (with names and parameters) backing the query.
+   *
+   * When set, `useQuery` will subscribe to those streams (and automatically handle unsubscribing from them, too).
+   *
+   * If {@link QuerySyncStreamOptions} is set on a stream, `useQuery` will remain in a loading state until that stream
+   * is marked as {@link SyncSubscriptionDescription.hasSynced}. This ensures the query is not missing rows that haven't
+   * been downloaded.
+   * Note however that after an initial sync, the query will not block itself while new rows are downloading. Instead,
+   * consistent sync snapshots will be made available as they've been processed by PowerSync.
+   **/
+  streams?: QuerySyncStreamOptions[];
 };
 
 export type UseBaseQueryOptions<TQueryOptions> = TQueryOptions & PowerSyncQueryOptions<any>;
@@ -120,14 +132,15 @@ function useQueryCore<
     throw new Error('PowerSync is not available');
   }
 
-  const { query, parameters, queryKey, ...resolvedOptions } = options;
+  const { query, parameters, queryKey, streams, ...resolvedOptions } = options;
 
-  const [{ queryFn }] = usePowerSyncQueries(
+  const { queries: [{ queryFn }], streamsHaveSynced } = usePowerSyncQueries(
     [
       {
         query,
         parameters,
-        queryKey
+        queryKey,
+        streams
       }
     ],
     queryClient
@@ -137,7 +150,8 @@ function useQueryCore<
     {
       ...(resolvedOptions as TQueryOptions),
       queryKey,
-      queryFn: query ? queryFn : resolvedOptions.queryFn
+      queryFn: query ? queryFn : resolvedOptions.queryFn,
+      enabled: streamsHaveSynced
     } as TQueryOptions,
     queryClient
   );