use a builder pattern to separate incremental watched query types.

stevensJourney · stevensJourney · commit c3e4709fcf55 · 2025-06-03T16:28:09.000+02:00
diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts
@@ -32,9 +32,9 @@ import {
   type PowerSyncConnectionOptions,
   type RequiredAdditionalConnectionOptions
 } 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';
+import { IncrementalWatchMode } from './watched/WatchedQueryBuilder.js';
+import { WatchedQueryBuilderMap } from './watched/WatchedQueryBuilderMap.js';
+import { WatchedQueryComparator } from './watched/processors/comparators.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -72,22 +72,6 @@ export interface PowerSyncDatabaseOptionsWithSettings extends BasePowerSyncDatab
   database: SQLOpenOptions;
 }
 
-export enum IncrementalWatchMode {
-  COMPARISON = 'comparison'
-}
-
-export interface WatchComparatorOptions<DataType> {
-  mode: IncrementalWatchMode.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[];
@@ -109,7 +93,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.
    */
-  processor?: WatchProcessorOptions<QueryResult>;
+  comparator?: WatchedQueryComparator<QueryResult>;
 }
 
 export interface WatchOnChangeEvent {
@@ -891,20 +875,14 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   }
 
   // TODO names
-  incrementalWatch<DataType>(options: IncrementalWatchOptions<DataType>): WatchedQuery<DataType> {
-    const { watch, processor } = options;
-
-    switch (options.processor?.mode) {
-      case IncrementalWatchMode.COMPARISON:
-      default:
-        return new OnChangeQueryProcessor({
-          db: this,
-          comparator: processor?.comparator ?? {
-            checkEquality: (a, b) => JSON.stringify(a) == JSON.stringify(b)
-          },
-          watchOptions: watch
-        });
+  incrementalWatch<Mode extends IncrementalWatchMode>(options: { mode: Mode }): WatchedQueryBuilderMap[Mode] {
+    const { mode } = options;
+    const builderFactory = WatchedQueryBuilderMap[mode];
+    if (!builderFactory) {
+      debugger;
+      throw new Error(`Unsupported watch mode: ${mode}`);
     }
+    return builderFactory(this);
   }
 
   /**
@@ -925,8 +903,11 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('onResult is required');
     }
 
-    // Uses shared incremental watch logic under the hook, but maintains the same external API as the old watch method.
-    const watchedQuery = this.incrementalWatch({
+    const { comparator } = options ?? {};
+    // This watch method which provides static SQL is currently only compatible with the comparison mode.
+    // Uses shared incremental watch logic under the hood, but maintains the same external API as the old watch method.
+    const watchedQuery = this.incrementalWatch({ mode: IncrementalWatchMode.COMPARISON }).build({
+      comparator,
       watch: {
         query: {
           compile: () => ({
@@ -938,10 +919,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
         placeholderData: null,
         reportFetching: false,
         throttleMs: options?.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS
-      },
-      processor: options?.processor ?? {
-        mode: IncrementalWatchMode.COMPARISON,
-        comparator: FalsyComparator
       }
     });
 
diff --git a/packages/common/src/client/watched/WatchedQueryBuilder.ts b/packages/common/src/client/watched/WatchedQueryBuilder.ts
@@ -0,0 +1,12 @@
+import { WatchedQuery } from './WatchedQuery.js';
+
+export enum IncrementalWatchMode {
+  COMPARISON = 'comparison'
+}
+
+/**
+ * Builds a {@link WatchedQuery} instance given a set of options.
+ */
+export interface WatchedQueryBuilder {
+  build<DataType>(options: {}): WatchedQuery<DataType>;
+}
diff --git a/packages/common/src/client/watched/WatchedQueryBuilderMap.ts b/packages/common/src/client/watched/WatchedQueryBuilderMap.ts
@@ -0,0 +1,17 @@
+import { AbstractPowerSyncDatabase } from '../AbstractPowerSyncDatabase.js';
+import { ComparisonWatchedQueryBuilder } from './processors/ComparisonWatchedQueryBuilder.js';
+import { IncrementalWatchMode } from './WatchedQueryBuilder.js';
+
+/**
+ * @internal
+ */
+export const WatchedQueryBuilderMap = {
+  [IncrementalWatchMode.COMPARISON]: (db: AbstractPowerSyncDatabase) => new ComparisonWatchedQueryBuilder(db)
+};
+
+/**
+ * @internal
+ */
+export type WatchedQueryBuilderMap = {
+  [key in IncrementalWatchMode]: ReturnType<(typeof WatchedQueryBuilderMap)[key]>;
+};
diff --git a/packages/common/src/client/watched/processors/ComparisonWatchedQueryBuilder.ts b/packages/common/src/client/watched/processors/ComparisonWatchedQueryBuilder.ts
@@ -0,0 +1,24 @@
+import { AbstractPowerSyncDatabase } from '../../../client/AbstractPowerSyncDatabase.js';
+import { WatchedQuery, WatchedQueryOptions } from '../WatchedQuery.js';
+import { WatchedQueryBuilder } from '../WatchedQueryBuilder.js';
+import { WatchedQueryComparator } from './comparators.js';
+import { OnChangeQueryProcessor } from './OnChangeQueryProcessor.js';
+
+export interface ComparisonWatchProcessorOptions<DataType> {
+  comparator?: WatchedQueryComparator<DataType>;
+  watch: WatchedQueryOptions<DataType>;
+}
+
+export class ComparisonWatchedQueryBuilder implements WatchedQueryBuilder {
+  constructor(protected db: AbstractPowerSyncDatabase) {}
+
+  build<DataType>(options: ComparisonWatchProcessorOptions<DataType>): WatchedQuery<DataType> {
+    return new OnChangeQueryProcessor({
+      db: this.db,
+      comparator: options.comparator ?? {
+        checkEquality: (a, b) => JSON.stringify(a) == JSON.stringify(b)
+      },
+      watchOptions: options.watch
+    });
+  }
+}
diff --git a/packages/common/src/client/watched/processors/OnChangeQueryProcessor.ts b/packages/common/src/client/watched/processors/OnChangeQueryProcessor.ts
@@ -1,9 +1,6 @@
 import { WatchedQueryState } from '../WatchedQuery.js';
 import { AbstractQueryProcessor, AbstractQueryProcessorOptions, LinkQueryOptions } from './AbstractQueryProcessor.js';
-
-export interface WatchedQueryComparator<Data> {
-  checkEquality: (current: Data, previous: Data) => boolean;
-}
+import { WatchedQueryComparator } from './comparators.js';
 
 /**
  * @internal
diff --git a/packages/common/src/client/watched/processors/comparators.ts b/packages/common/src/client/watched/processors/comparators.ts
@@ -1,4 +1,6 @@
-import { WatchedQueryComparator } from './OnChangeQueryProcessor.js';
+export interface WatchedQueryComparator<Data> {
+  checkEquality: (current: Data, previous: Data) => boolean;
+}
 
 export type ArrayComparatorOptions<ItemType> = {
   compareBy: (item: ItemType) => string;
diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts
@@ -34,6 +34,7 @@ export * from './client/watched/GetAllQuery.js';
 export * from './client/watched/processors/AbstractQueryProcessor.js';
 export * from './client/watched/processors/comparators.js';
 export * from './client/watched/WatchedQuery.js';
+export * from './client/watched/WatchedQueryBuilder.js';
 
 export * from './utils/AbortOperation.js';
 export * from './utils/BaseObserver.js';
diff --git a/packages/react/README.md b/packages/react/README.md
@@ -345,12 +345,9 @@ function MyWidget() {
   // Note that isFetching is set (by default) whenever the query is being fetched/checked.
   // This will result in `MyWidget` re-rendering for any change to the `cats` table.
   const { data, isLoading, isFetching } = useQuery(`SELECT * FROM cats WHERE breed = 'tabby'`, [], {
-    processor: {
-      mode: IncrementalWatchMode.COMPARISON,
       comparator: new ArrayComparator({
         compareBy: (cat) => JSON.stringify(cat)
       })
-    },
   })
 
   // ... Widget code
@@ -375,12 +372,8 @@ function MyWidget() {
   // When reportFetching == false the object returned from useQuery will only be changed when the data, isLoading or error state changes.
   // This method performs a comparison in memory in order to determine changes.
   const { data, isLoading } = useQuery(`SELECT * FROM cats WHERE breed = 'tabby'`, [], {
-    processor: {
-      mode: 'comparison',
-      comparator: new ArrayComparator({
-        compareBy: (cat) => JSON.stringify(cat)
-      })
-    },
+    comparator: new ArrayComparator({
+      compareBy: (cat) => JSON.stringify(cat)
     reportFetching: false
   })
 
diff --git a/packages/react/src/QueryStore.ts b/packages/react/src/QueryStore.ts
@@ -1,4 +1,4 @@
-import { AbstractPowerSyncDatabase, WatchCompatibleQuery, WatchedQuery } from '@powersync/common';
+import { AbstractPowerSyncDatabase, IncrementalWatchMode, WatchCompatibleQuery, WatchedQuery } from '@powersync/common';
 import { AdditionalOptions } from './hooks/watched/watch-types';
 
 export function generateQueryKey(
@@ -19,14 +19,18 @@ export class QueryStore {
       return this.cache.get(key);
     }
 
-    const watchedQuery = this.db.incrementalWatch({
-      watch: {
-        query,
-        placeholderData: [],
-        throttleMs: options.throttleMs
-      },
-      processor: options.processor
-    });
+    const watchedQuery = this.db
+      .incrementalWatch({
+        mode: IncrementalWatchMode.COMPARISON
+      })
+      .build({
+        watch: {
+          query,
+          placeholderData: [],
+          throttleMs: options.throttleMs
+        },
+        comparator: options.comparator
+      });
 
     const disposer = watchedQuery.registerListener({
       closed: () => {
diff --git a/packages/react/src/hooks/suspense/useSuspenseQuery.ts b/packages/react/src/hooks/suspense/useSuspenseQuery.ts
@@ -1,4 +1,4 @@
-import { CompilableQuery, FalsyComparator, IncrementalWatchMode } from '@powersync/common';
+import { CompilableQuery, FalsyComparator } from '@powersync/common';
 import { AdditionalOptions } from '../watched/watch-types';
 import { SuspenseQueryResult } from './SuspenseQueryResult';
 import { useSingleSuspenseQuery } from './useSingleSuspenseQuery';
@@ -36,10 +36,8 @@ export const useSuspenseQuery = <T = any>(
     default:
       return useWatchedSuspenseQuery<T>(query, parameters, {
         ...options,
-        processor: options.processor ?? {
-          mode: IncrementalWatchMode.COMPARISON,
-          comparator: FalsyComparator // Default comparator that always reports changed result sets
-        }
+        // Default comparator that always reports changed result sets
+        comparator: options.comparator ?? FalsyComparator
       });
   }
 };
diff --git a/packages/react/src/hooks/watched/useQuery.ts b/packages/react/src/hooks/watched/useQuery.ts
@@ -1,4 +1,4 @@
-import { FalsyComparator, IncrementalWatchMode, type CompilableQuery } from '@powersync/common';
+import { FalsyComparator, type CompilableQuery } from '@powersync/common';
 import { usePowerSync } from '../PowerSyncContext';
 import { useSingleQuery } from './useSingleQuery';
 import { useWatchedQuery } from './useWatchedQuery';
@@ -43,14 +43,10 @@ export const useQuery = <RowType = any>(
         powerSync,
         queryChanged,
         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: IncrementalWatchMode.COMPARISON,
-            comparator: FalsyComparator
-          }
+          // Maintains backwards compatibility with previous versions
+          // Comparisons are opt-in by default
+          // We emit new data for each table change by default.
+          comparator: options.comparator ?? FalsyComparator
         }
       });
   }
diff --git a/packages/react/src/hooks/watched/useWatchedQuery.ts b/packages/react/src/hooks/watched/useWatchedQuery.ts
@@ -1,3 +1,4 @@
+import { FalsyComparator, IncrementalWatchMode } from '@powersync/common';
 import React from 'react';
 import { HookWatchOptions, QueryResult } from './watch-types';
 import { InternalHookOptions } from './watch-utils';
@@ -8,14 +9,14 @@ export const useWatchedQuery = <RowType = unknown>(
   const { query, powerSync, queryChanged, options: hookOptions } = options;
 
   const createWatchedQuery = React.useCallback(() => {
-    return powerSync.incrementalWatch<RowType[]>({
+    return powerSync.incrementalWatch({ mode: IncrementalWatchMode.COMPARISON }).build<RowType[]>({
       watch: {
         placeholderData: [],
         query,
         throttleMs: hookOptions.throttleMs,
         reportFetching: hookOptions.reportFetching
       },
-      processor: hookOptions.processor
+      comparator: hookOptions.comparator ?? FalsyComparator
     });
   }, []);
 
diff --git a/packages/react/src/hooks/watched/watch-types.ts b/packages/react/src/hooks/watched/watch-types.ts
@@ -1,8 +1,8 @@
-import { WatchProcessorOptions, type SQLWatchOptions } from '@powersync/common';
+import { WatchedQueryComparator, type SQLWatchOptions } from '@powersync/common';
 
-export interface HookWatchOptions<RowType = unknown> extends Omit<SQLWatchOptions, 'signal' | 'processor'> {
+export interface HookWatchOptions<RowType = unknown> extends Omit<SQLWatchOptions, 'signal' | 'comparator'> {
   reportFetching?: boolean;
-  processor?: WatchProcessorOptions<RowType[]>;
+  comparator?: WatchedQueryComparator<RowType[]>;
 }
 
 export interface AdditionalOptions<RowType = unknown> extends HookWatchOptions<RowType> {
diff --git a/packages/react/tests/useQuery.test.tsx b/packages/react/tests/useQuery.test.tsx
@@ -202,12 +202,9 @@ describe('useQuery', () => {
     const { result } = renderHook(
       () =>
         useQuery('SELECT * FROM lists WHERE name = ?', ['aname'], {
-          processor: {
-            mode: 'comparison',
-            comparator: new commonSdk.ArrayComparator({
-              compareBy: (item) => JSON.stringify(item)
-            })
-          }
+          comparator: new commonSdk.ArrayComparator({
+            compareBy: (item) => JSON.stringify(item)
+          })
         }),
       { wrapper }
     );
@@ -317,7 +314,7 @@ describe('useQuery', () => {
     // This query can be instantiated once and reused.
     // The query retains it's state and will not re-fetch the data unless the result changes.
     // This is useful for queries that are used in multiple components.
-    const listsQuery = db.incrementalWatch({
+    const listsQuery = db.incrementalWatch({ mode: commonSdk.IncrementalWatchMode.COMPARISON }).build({
       watch: {
         placeholderData: [],
         query: {
diff --git a/packages/react/tests/useSuspenseQuery.test.tsx b/packages/react/tests/useSuspenseQuery.test.tsx
@@ -1,4 +1,4 @@
-import { AbstractPowerSyncDatabase, WatchedQuery } from '@powersync/common';
+import { AbstractPowerSyncDatabase, IncrementalWatchMode, WatchedQuery } from '@powersync/common';
 import { cleanup, renderHook, screen, waitFor } from '@testing-library/react';
 import React from 'react';
 import { ErrorBoundary } from 'react-error-boundary';
@@ -64,10 +64,22 @@ describe('useSuspenseQuery', () => {
   it('should suspend on initial load', async () => {
     // spy on watched query generation
     const baseImplementation = powersync.incrementalWatch;
-    let watch: WatchedQuery<Array<any>> | null = null;
+    let watch: WatchedQuery<any> | null = null;
+
     const spy = vi.spyOn(powersync, 'incrementalWatch').mockImplementation((options) => {
-      watch = baseImplementation.call(powersync, options);
-      return watch!;
+      if (options.mode !== IncrementalWatchMode.COMPARISON) {
+        return baseImplementation.call(powersync, options);
+      }
+
+      const builder = baseImplementation.call(powersync, options);
+      const baseBuild = builder.build;
+
+      vi.spyOn(builder, 'build').mockImplementation((buildOptions) => {
+        watch = baseBuild.call(builder, buildOptions);
+        return watch!;
+      });
+
+      return builder!;
     });
 
     const wrapper = ({ children }) => (
@@ -250,7 +262,7 @@ describe('useSuspenseQuery', () => {
     // This query can be instantiated once and reused.
     // The query retains it's state and will not re-fetch the data unless the result changes.
     // This is useful for queries that are used in multiple components.
-    const listsQuery = db.incrementalWatch({
+    const listsQuery = db.incrementalWatch({ mode: IncrementalWatchMode.COMPARISON }).build({
       watch: {
         placeholderData: [],
         query: {
diff --git a/packages/react/vitest.config.ts b/packages/react/vitest.config.ts
@@ -32,7 +32,7 @@ const config: UserConfigExport = {
        */
       isolate: true,
       provider: 'playwright',
-      headless: true,
+      headless: false,
       instances: [
         {
           browser: 'chromium'
diff --git a/packages/vue/src/composables/useSingleQuery.ts b/packages/vue/src/composables/useSingleQuery.ts
diff --git a/packages/vue/src/composables/useWatchedQuery.ts b/packages/vue/src/composables/useWatchedQuery.ts
diff --git a/packages/web/tests/watch.test.ts b/packages/web/tests/watch.test.ts
