maintain backwards compatibility

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -34,6 +34,7 @@ import {
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
 import { WatchedQuery, WatchedQueryOptions } from './watched/WatchedQuery.js';
 import { OnChangeQueryProcessor, WatchedQueryComparator } from './watched/processors/OnChangeQueryProcessor.js';
+import { FalsyComparator } from './watched/processors/comparators.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -71,6 +72,18 @@ export interface PowerSyncDatabaseOptionsWithSettings extends BasePowerSyncDatab
   database: SQLOpenOptions;
 }
 
+export interface WatchComparatorOptions<DataType> {
+  mode: 'comparison';
+  comparator?: WatchedQueryComparator<DataType>;
+}
+
+export type WatchProcessorOptions<DataType> = WatchComparatorOptions<DataType>;
+
+export interface IncrementalWatchOptions<DataType> {
+  watch: WatchedQueryOptions<DataType>;
+  processor?: WatchProcessorOptions<DataType>;
+}
+
 export interface SQLWatchOptions {
   signal?: AbortSignal;
   tables?: string[];
@@ -92,7 +105,7 @@ export interface SQLWatchOptions {
    * Optional comparator which will be used to compare the results of the query.
    * The watched query will only yield results if the comparator returns false.
    */
-  comparator?: WatchedQueryComparator<QueryResult>;
+  processor?: WatchProcessorOptions<QueryResult>;
 }
 
 export interface WatchOnChangeEvent {
@@ -109,16 +122,6 @@ export interface WatchOnChangeHandler {
   onError?: (error: Error) => void;
 }
 
-export interface ComparatorWatchOptions<DataType> {
-  mode: 'comparison';
-  comparator?: WatchedQueryComparator<DataType>;
-}
-
-export interface IncrementalWatchOptions<DataType> {
-  watch: WatchedQueryOptions<DataType>;
-  processor?: ComparatorWatchOptions<DataType>;
-}
-
 export interface PowerSyncDBListener extends StreamingSyncImplementationListener {
   initialized: () => void;
   schemaChanged: (schema: Schema) => void;
@@ -916,25 +919,27 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('onResult is required');
     }
 
-    const watch = new OnChangeQueryProcessor({
-      db: this,
-      // Comparisons are disabled if no comparator is provided
-      comparator: options?.comparator,
-      watchOptions: {
-        placeholderData: null,
+    // Uses shared incremental watch logic under the hook, but maintains the same external API as the old watch method.
+    const watchedQuery = this.incrementalWatch({
+      watch: {
         query: {
           compile: () => ({
             sql: sql,
             parameters: parameters ?? []
           }),
           execute: () => this.executeReadOnly(sql, parameters)
         },
-        throttleMs: options?.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS,
-        reportFetching: false
+        placeholderData: null,
+        reportFetching: false,
+        throttleMs: options?.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS
+      },
+      processor: options?.processor ?? {
+        mode: 'comparison',
+        comparator: FalsyComparator
       }
     });
 
-    const dispose = watch.subscribe({
+    const dispose = watchedQuery.subscribe({
       onData: (data) => {
         if (!data) {
           // This should not happen. We only use null for the initial data.
@@ -949,7 +954,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
     options?.signal?.addEventListener('abort', () => {
       dispose();
-      watch.close();
+      watchedQuery.close();
     });
   }
 

packages/common/src/client/watched/processors/comparators.ts

@@ -0,0 +1,43 @@
+import { WatchedQueryComparator } from './OnChangeQueryProcessor.js';
+
+export type ArrayComparatorOptions<ItemType> = {
+  compareBy: (item: ItemType) => string;
+};
+
+/**
+ * Compares array results of watched queries.
+ */
+export class ArrayComparator<ItemType> implements WatchedQueryComparator<ItemType[]> {
+  constructor(protected options: ArrayComparatorOptions<ItemType>) {}
+
+  checkEquality(current: ItemType[], previous: ItemType[]) {
+    if (current.length === 0 && previous.length === 0) {
+      return true;
+    }
+
+    if (current.length !== previous.length) {
+      return false;
+    }
+
+    const { compareBy } = this.options;
+
+    // At this point the lengths are equal
+    for (let i = 0; i < current.length; i++) {
+      const currentItem = compareBy(current[i]);
+      const previousItem = compareBy(previous[i]);
+
+      if (currentItem !== previousItem) {
+        return false;
+      }
+    }
+
+    return true;
+  }
+}
+
+/**
+ * Watched query comparator that always reports changed result sets.
+ */
+export const FalsyComparator: WatchedQueryComparator<unknown> = {
+  checkEquality: () => false // Default comparator that always returns false
+};

packages/common/src/index.ts

@@ -30,7 +30,8 @@ export * from './db/schema/Schema.js';
 export * from './db/schema/Table.js';
 export * from './db/schema/TableV2.js';
 
-// TODO other exports
+export * from './client/watched/processors/AbstractQueryProcessor.js';
+export * from './client/watched/processors/comparators.js';
 export * from './client/watched/WatchedQuery.js';
 
 export * from './utils/AbortOperation.js';

packages/react/src/QueryStore.ts

@@ -24,7 +24,8 @@ export class QueryStore {
         query,
         placeholderData: [],
         throttleMs: options.throttleMs
-      }
+      },
+      processor: options.processor
     });
 
     const disposer = watchedQuery.registerListener({

packages/react/src/hooks/suspense/useSuspenseQuery.ts

@@ -1,4 +1,4 @@
-import { CompilableQuery } from '@powersync/common';
+import { CompilableQuery, FalsyComparator } from '@powersync/common';
 import { AdditionalOptions } from '../watched/watch-types';
 import { SuspenseQueryResult } from './SuspenseQueryResult';
 import { useSingleSuspenseQuery } from './useSingleSuspenseQuery';
@@ -28,12 +28,18 @@ import { useWatchedSuspenseQuery } from './useWatchedSuspenseQuery';
 export const useSuspenseQuery = <T = any>(
   query: string | CompilableQuery<T>,
   parameters: any[] = [],
-  options: AdditionalOptions = {}
+  options: AdditionalOptions<T> = {}
 ): SuspenseQueryResult<T> => {
   switch (options.runQueryOnce) {
     case true:
       return useSingleSuspenseQuery<T>(query, parameters, options);
     default:
-      return useWatchedSuspenseQuery<T>(query, parameters, options);
+      return useWatchedSuspenseQuery<T>(query, parameters, {
+        ...options,
+        processor: options.processor ?? {
+          mode: 'comparison',
+          comparator: FalsyComparator // Default comparator that always reports changed result sets
+        }
+      });
   }
 };

packages/react/src/hooks/watched/useQuery.ts

@@ -1,4 +1,4 @@
-import { type CompilableQuery } from '@powersync/common';
+import { FalsyComparator, type CompilableQuery } from '@powersync/common';
 import { usePowerSync } from '../PowerSyncContext';
 import { useSingleQuery } from './useSingleQuery';
 import { useWatchedQuery } from './useWatchedQuery';
@@ -21,7 +21,7 @@ import { constructCompatibleQuery } from './watch-utils';
 export const useQuery = <RowType = any>(
   query: string | CompilableQuery<RowType>,
   parameters: any[] = [],
-  options: AdditionalOptions = { runQueryOnce: false }
+  options: AdditionalOptions<RowType> = { runQueryOnce: false }
 ): QueryResult<RowType> => {
   const powerSync = usePowerSync();
   if (!powerSync) {
@@ -42,7 +42,16 @@ export const useQuery = <RowType = any>(
         query: parsedQuery,
         powerSync,
         queryChanged,
-        options
+        options: {
+          ...options,
+          processor: options.processor ?? {
+            // Maintains backwards compatibility with previous versions
+            // Comparisons are opt-in by default
+            // We emit new data for each table change by default.
+            mode: 'comparison',
+            comparator: FalsyComparator
+          }
+        }
       });
   }
 };

packages/react/src/hooks/watched/useWatchedQuery.ts

@@ -15,7 +15,8 @@ export const useWatchedQuery = <RowType = unknown>(
         query,
         throttleMs: hookOptions.throttleMs,
         reportFetching: hookOptions.reportFetching
-      }
+      },
+      processor: hookOptions.processor
     });
   }, []);
 

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

@@ -1,10 +1,11 @@
-import { type SQLWatchOptions } from '@powersync/common';
+import { WatchProcessorOptions, type SQLWatchOptions } from '@powersync/common';
 
-export interface HookWatchOptions extends Omit<SQLWatchOptions, 'signal'> {
+export interface HookWatchOptions<RowType = unknown> extends Omit<SQLWatchOptions, 'signal' | 'processor'> {
   reportFetching?: boolean;
+  processor?: WatchProcessorOptions<RowType[]>;
 }
 
-export interface AdditionalOptions extends HookWatchOptions {
+export interface AdditionalOptions<RowType = unknown> extends HookWatchOptions<RowType> {
   runQueryOnce?: boolean;
 }
 

packages/react/tests/useQuery.test.tsx

@@ -1,4 +1,3 @@
-import React from 'react';
 import * as commonSdk from '@powersync/common';
 import { PowerSyncDatabase } from '@powersync/web';
 import { act, cleanup, renderHook, waitFor } from '@testing-library/react';
@@ -195,7 +194,18 @@ describe('useQuery', () => {
   it('should emit result data when query changes', async () => {
     const db = openPowerSync();
     const wrapper = ({ children }) => <PowerSyncContext.Provider value={db}>{children}</PowerSyncContext.Provider>;
-    const { result } = renderHook(() => useQuery('SELECT * FROM lists WHERE name = ?', ['aname']), { wrapper });
+    const { result } = renderHook(
+      () =>
+        useQuery('SELECT * FROM lists WHERE name = ?', ['aname'], {
+          processor: {
+            mode: 'comparison',
+            comparator: new commonSdk.ArrayComparator({
+              compareBy: (item) => JSON.stringify(item)
+            })
+          }
+        }),
+      { wrapper }
+    );
 
     expect(result.current.isLoading).toEqual(true);
 
@@ -256,6 +266,46 @@ describe('useQuery', () => {
     expect(data == result.current.data).toEqual(true);
   });
 
+  // Verifies backwards compatibility with the previous implementation (no comparison)
+  it('should emit result data when data changes when not using comparator', async () => {
+    const db = openPowerSync();
+    const wrapper = ({ children }) => <PowerSyncContext.Provider value={db}>{children}</PowerSyncContext.Provider>;
+    const { result } = renderHook(() => useQuery('SELECT * FROM lists WHERE name = ?', ['aname']), { wrapper });
+
+    expect(result.current.isLoading).toEqual(true);
+
+    await waitFor(
+      async () => {
+        const { current } = result;
+        expect(current.isLoading).toEqual(false);
+      },
+      { timeout: 500, interval: 100 }
+    );
+
+    // This should trigger an update
+    await db.execute('INSERT INTO lists(id, name) VALUES (uuid(), ?)', ['aname']);
+
+    // Keep track of the previous data reference
+    let previousData = result.current.data;
+    await waitFor(
+      async () => {
+        const { current } = result;
+        expect(current.data.length).toEqual(1);
+        previousData = current.data;
+      },
+      { timeout: 500, interval: 100 }
+    );
+
+    // This should still trigger an update since the underlaying tables changed.
+    await db.execute('INSERT INTO lists(id, name) VALUES (uuid(), ?)', ['noname']);
+
+    // It's difficult to assert no update happened, but we can wait a bit
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+
+    // It should be the same data array reference, no update should have happened
+    expect(result.current.data == previousData).false;
+  });
+
   it('should use an existing WatchedQuery instance', async () => {
     const db = openPowerSync();