Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions packages/tanstack-react-query/src/hooks/usePowerSyncQueries.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { type CompilableQuery, parseQuery } from '@powersync/common';
import { QuerySyncStreamOptions, useAllSyncStreamsHaveSynced, usePowerSync } from '@powersync/react';
import { useEffect, useState, useCallback, useMemo } from 'react';
import { useEffect, useState, useCallback, useMemo, useRef } from 'react';
import * as Tanstack from '@tanstack/react-query';

export type UsePowerSyncQueriesInput = {
Expand Down Expand Up @@ -111,7 +111,17 @@ export function usePowerSyncQueries(
}))
);

// Tracks the []->[tables] transition to rescue data lost during pending resolveTables.
const tablesInitialized = useRef<boolean[]>([]);

if (tablesInitialized.current.length !== parsedQueries.length) {
tablesInitialized.current = parsedQueries.map(() => false);
}

useEffect(() => {
// Re-arm rescue tracking when queries change.
tablesInitialized.current = parsedQueries.map(() => false);

const listeners = parsedQueries.map((pq, idx) => {
if (pq.parseError || !pq.query) {
return null;
Expand Down Expand Up @@ -154,6 +164,12 @@ export function usePowerSyncQueries(

const abort = new AbortController();

// Rescue data lost while resolveTables was pending (listener watched [] tables).
if (tablesArr[idx]?.length > 0 && !tablesInitialized.current[idx]) {
tablesInitialized.current[idx] = true;
queryClient.invalidateQueries({ queryKey: pq.queryKey });
}

powerSync.onChangeWithCallback(
{
onChange: () => {
Expand Down Expand Up @@ -203,5 +219,5 @@ export function usePowerSyncQueries(
}),
streamsHaveSynced
};
}, [parsedQueries, errorsArr, tablesArr, powerSync]);
}, [parsedQueries, errorsArr, tablesArr, powerSync, streamsHaveSynced]);
}
8 changes: 4 additions & 4 deletions packages/tanstack-react-query/src/hooks/useQueries.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
{
Expand Down
9 changes: 6 additions & 3 deletions packages/tanstack-react-query/src/hooks/useQuery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ function useQueryCore<
const { query, parameters, queryKey, streams, ...resolvedOptions } = options;

const {
queries: [{ queryFn }],
queries: [{ queryFn: powerSyncQueryFn }],
streamsHaveSynced
} = usePowerSyncQueries(
[
Expand All @@ -149,12 +149,15 @@ function useQueryCore<
queryClient
);

const isSuspense = useQueryFn === 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
);
Expand Down
100 changes: 100 additions & 0 deletions packages/tanstack-react-query/tests/enabled.test.tsx
Original file line number Diff line number Diff line change
@@ -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');
});
});
143 changes: 143 additions & 0 deletions packages/tanstack-react-query/tests/usePowerSyncQueries.test.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
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('Bug 1: streamsHaveSynced in final useMemo deps', () => {
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();
});
});

describe('Bug 2: first table-resolution race loses first-sync data', () => {
it('reflects rows written while table resolution is still on the slow path', 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();
});
});
});