Don't wait for streams by default

Author: simolus3

packages/react/src/hooks/streams.ts

@@ -8,6 +8,7 @@ import {
   SyncStreamSubscription
 } from '@powersync/common';
 import { useStatus } from './useStatus.js';
+import { QuerySyncStreamOptions } from './watched/watch-types.js';
 
 /**
  * A sync stream to subscribe to in {@link useSyncStream}.
@@ -51,12 +52,15 @@ export function useSyncStream(options: UseSyncStreamOptions): SyncStreamStatus |
           subscription = sub;
           setSubscription(sub);
         } else {
+          // The cleanup function already ran, unsubscribe immediately.
           sub.unsubscribe();
         }
       });
 
     return () => {
       active = false;
+      // If we don't have a subscription yet, it'll still get cleaned up once the promise resolves because we've set
+      // active to false.
       subscription?.unsubscribe();
     };
   }, [name, parameters]);
@@ -69,12 +73,12 @@ export function useSyncStream(options: UseSyncStreamOptions): SyncStreamStatus |
  */
 export function useAllSyncStreamsHaveSynced(
   db: AbstractPowerSyncDatabase,
-  streams: UseSyncStreamOptions[] | undefined
+  streams: QuerySyncStreamOptions[] | undefined
 ): boolean {
   // Since streams are a user-supplied array, they will likely be different each time this function is called. We don't
   // want to update underlying subscriptions each time, though.
   const hash = useMemo(() => streams && JSON.stringify(streams), [streams]);
-  const [synced, setHasSynced] = useState(streams == null || streams.length == 0);
+  const [synced, setHasSynced] = useState(streams == null || streams.every((e) => e.waitForStream != true));
 
   useEffect(() => {
     if (streams) {
@@ -87,9 +91,12 @@ export function useAllSyncStreamsHaveSynced(
       }
 
       // First, wait for all subscribe() calls to make all subscriptions active.
-      Promise.all(promises).then(async (streams) => {
+      Promise.all(promises).then(async (resolvedStreams) => {
         function allHaveSynced(status: SyncStatus) {
-          return streams.every((s) => status.forStream(s)?.subscription?.hasSynced);
+          return resolvedStreams.every((s, i) => {
+            const request = streams[i];
+            return !request.waitForStream || status.forStream(s)?.subscription?.hasSynced;
+          });
         }
 
         // Wait for the effect to be cancelled or all streams having synced.
@@ -107,7 +114,7 @@ export function useAllSyncStreamsHaveSynced(
         }
 
         // Effect was definitely cancelled at this point, so drop the subscriptions.
-        for (const stream of streams) {
+        for (const stream of resolvedStreams) {
           stream.unsubscribe();
         }
       });

packages/react/src/hooks/watched/watch-types.ts

@@ -6,18 +6,30 @@ export interface HookWatchOptions extends Omit<SQLOnChangeOptions, 'signal'> {
    * 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).
-   * Additionally, `useQuery` will remainin in a loading state until all of the streams are marked as
-   * {@link SyncSubscriptionDescription.hasSynced}. This ensures the query is not missing rows that haven't been
-   * downloaded.
+   *
+   * 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.
    *
    * @experimental Sync streams are currently in alpha.
    */
-  streams?: UseSyncStreamOptions[];
+  streams?: QuerySyncStreamOptions[];
   reportFetching?: boolean;
 }
 
+/**
+ * Additional options to control how `useQuery` behaves when subscribing to a stream.
+ */
+export interface QuerySyncStreamOptions extends UseSyncStreamOptions {
+  /**
+   * When set to `true`, a `useQuery` hook will remain in a loading state as long as the stream is resolving or
+   * downloading for the first time (in other words, until {@link SyncSubscriptionDescription.hasSynced} is true).
+   */
+  waitForStream?: boolean;
+}
+
 export interface AdditionalOptions extends HookWatchOptions {
   runQueryOnce?: boolean;
 }

packages/react/tests/streams.test.tsx

@@ -6,6 +6,7 @@ import { openPowerSync } from './utils';
 import { PowerSyncContext } from '../src/hooks/PowerSyncContext';
 import { useSyncStream, UseSyncStreamOptions } from '../src/hooks/streams';
 import { useQuery } from '../src/hooks/watched/useQuery';
+import { QuerySyncStreamOptions } from '../src/hooks/watched/watch-types';
 
 describe('stream hooks', () => {
   let db: AbstractPowerSyncDatabase;
@@ -51,10 +52,13 @@ describe('stream hooks', () => {
         expect(currentStreams()).toStrictEqual([]);
       });
 
-      it('useQuery', async () => {
-        const { result } = renderHook(() => useQuery('SELECT 1', [], { streams: [{ name: 'a' }] }), {
-          wrapper: testWrapper
-        });
+      it('useQuery waiting on stream', async () => {
+        const { result } = renderHook(
+          () => useQuery('SELECT 1', [], { streams: [{ name: 'a', waitForStream: true }] }),
+          {
+            wrapper: testWrapper
+          }
+        );
         expect(result.current).toMatchObject({ isLoading: true });
         // Including the stream should subscribe.
         await waitFor(() => expect(currentStreams()).toHaveLength(1), { timeout: 1000, interval: 100 });
@@ -68,6 +72,16 @@ describe('stream hooks', () => {
         await waitFor(() => expect(result.current.data).toHaveLength(1), { timeout: 1000, interval: 100 });
       });
 
+      it('useQuery not waiting on stream', async () => {
+        // By default, it should still run the query immediately instead of waiting for the stream to resolve.
+        const { result } = renderHook(() => useQuery('SELECT 1', [], { streams: [{ name: 'a' }] }), {
+          wrapper: testWrapper
+        });
+
+        // Not resolving the subscription.
+        await waitFor(() => expect(result.current.data).toHaveLength(1), { timeout: 1000, interval: 100 });
+      });
+
       it('unsubscribes on unmount', async () => {
         const { unmount } = renderHook(() => useQuery('SELECT 1', [], { streams: [{ name: 'a' }, { name: 'b' }] }), {
           wrapper: testWrapper
@@ -80,7 +94,7 @@ describe('stream hooks', () => {
 
       it('handles stream parameter changes', async () => {
         // Start without streams
-        let streams: UseSyncStreamOptions[] = [];
+        let streams: QuerySyncStreamOptions[] = [];
         let streamUpdateListeners: (() => void)[] = [];
 
         const { result } = renderHook(
@@ -108,7 +122,7 @@ describe('stream hooks', () => {
         await waitFor(() => expect(result.current.data).toHaveLength(1), { timeout: 1000, interval: 100 });
 
         // Adopt streams - this should reset back to loading
-        streams = [{ name: 'a' }];
+        streams = [{ name: 'a', waitForStream: true }];
         act(() => streamUpdateListeners.forEach((cb) => cb()));
         expect(result.current).toMatchObject({ isLoading: true });