Invert Watched Query creation API. Update hook packages to provide differentiator implementation.

Author: stevensJourney

.changeset/little-bananas-fetch.md

@@ -3,5 +3,5 @@
 ---
 
 - Added additional listeners for `closing` and `closed` events in `AbstractPowerSyncDatabase`.
-- Added `incrementalWatch` API for enhanced watched queries.
+- Added `query` and `customQuery` APIs for enhanced watched queries.
 - Added `triggerImmediate` option to the `onChange` API. This allows emitting an initial event which can be useful for downstream use cases.

demos/yjs-react-supabase-text-collab/README.md

@@ -44,7 +44,7 @@ docker run \
 -p 8080:8080 \
 -e POWERSYNC_CONFIG_B64=$(base64 -i ./powersync.yaml) \
 -e POWERSYNC_SYNC_RULES_B64=$(base64 -i ./sync-rules.yaml) \
---env-file ./.env \
+--env-file ./.env.local \
 --network supabase_network_yjs-react-supabase-text-collab \
 --name my-powersync journeyapps/powersync-service:latest
 ```

demos/yjs-react-supabase-text-collab/src/library/powersync/PowerSyncYjsProvider.ts

@@ -1,7 +1,7 @@
 import * as Y from 'yjs';
 
 import { b64ToUint8Array, Uint8ArrayTob64 } from '@/library/binary-utils';
-import { AbstractPowerSyncDatabase, GetAllQuery, IncrementalWatchMode } from '@powersync/web';
+import { AbstractPowerSyncDatabase } from '@powersync/web';
 import { ObservableV2 } from 'lib0/observable';
 import { v4 as uuidv4 } from 'uuid';
 import { DocumentUpdates } from './AppSchema';
@@ -41,22 +41,20 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
      * This will be used to apply updates from other editors.
      * When we received an added item we apply the update to the Yjs document.
      */
-    const updateQuery = db.incrementalWatch({ mode: IncrementalWatchMode.DIFFERENTIAL }).build({
-      watch: {
-        query: new GetAllQuery<DocumentUpdates>({
-          sql: /* sql */ `
-            SELECT
-              *
-            FROM
-              document_updates
-            WHERE
-              document_id = ?
-              AND editor_id != ?
-          `,
-          parameters: [documentId, this.id]
-        })
-      }
-    });
+    const updateQuery = db
+      .query<DocumentUpdates>({
+        sql: /* sql */ `
+          SELECT
+            *
+          FROM
+            document_updates
+          WHERE
+            document_id = ?
+            AND editor_id != ?
+        `,
+        parameters: [documentId, this.id]
+      })
+      .differentialWatch();
 
     this.abortController.signal.addEventListener(
       'abort',
@@ -73,8 +71,8 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
     let synced = false;
 
     updateQuery.registerListener({
-      onData: async (diff) => {
-        for (const added of diff.added) {
+      onStateChange: async () => {
+        for (const added of updateQuery.state.diff.added) {
           Y.applyUpdateV2(doc, b64ToUint8Array(added.update_b64));
         }
         if (!synced) {

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -18,6 +18,8 @@ import { ControlledExecutor } from '../utils/ControlledExecutor.js';
 import { throttleTrailing } from '../utils/async.js';
 import { mutexRunExclusive } from '../utils/mutex.js';
 import { ConnectionManager } from './ConnectionManager.js';
+import { CustomQuery } from './CustomQuery.js';
+import { ArrayQueryDefinition, Query } from './Query.js';
 import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js';
 import { BucketStorageAdapter, PSInternalTable } from './sync/bucket/BucketStorageAdapter.js';
@@ -33,9 +35,9 @@ import {
   type PowerSyncConnectionOptions,
   type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
-import { IncrementalWatchMode } from './watched/WatchedQueryBuilder.js';
-import { WatchedQueryBuilderMap } from './watched/WatchedQueryBuilderMap.js';
-import { FalsyComparator, WatchedQueryComparator } from './watched/processors/comparators.js';
+import { DEFAULT_WATCH_THROTTLE_MS, WatchCompatibleQuery } from './watched/WatchedQuery.js';
+import { OnChangeQueryProcessor } from './watched/processors/OnChangeQueryProcessor.js';
+import { WatchedQueryComparator } from './watched/processors/comparators.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -139,8 +141,6 @@ export const DEFAULT_POWERSYNC_CLOSE_OPTIONS: PowerSyncCloseOptions = {
   disconnect: true
 };
 
-export const DEFAULT_WATCH_THROTTLE_MS = 30;
-
 export const DEFAULT_POWERSYNC_DB_OPTIONS = {
   retryDelayMs: 5000,
   logger: Logger.get('PowerSyncDatabase'),
@@ -890,41 +890,59 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   }
 
   /**
-   * Watch a SQL query which incrementally emits updates for result sets.
-   * This is useful for only getting updates when the result set changes, or viewing the change in the result set over time.
-   * @returns A {@link WatchedQueryBuilder} for the specified watch mode.
+   * Allows defining a query which can be used to build a {@link WatchedQuery}.
+   * The defined query will be executed with {@link AbstractPowerSyncDatabase.getAll}.
+   * An optional mapper function can be provided to transform the results.
    *
-   * For a comparison based watch, use {@link IncrementalWatchMode.COMPARISON}.
-   * See {@link ComparisonWatchedQueryBuilder} for more details.
    * @example
    * ```javascript
-   * const watchedQuery = powerSync
-   *  .incrementalWatch({ mode: IncrementalWatchMode.COMPARISON })
-   *  .build({
-   *    // ... Options
-   *  })
+   * const watchedTodos = powersync.query({
+   *    sql: `SELECT photo_id as id FROM todos WHERE photo_id IS NOT NULL`,
+   *    parameters: [],
+   *    mapper: (row) => ({
+   *      ...row,
+   *      created_at: new Date(row.created_at as string)
+   *    })
+   * })
+   * .watch()
+   * // OR use .differentialWatch() for fine-grained watches.
    * ```
+   */
+  query<RowType>(query: ArrayQueryDefinition<RowType>): Query<RowType> {
+    const { sql, parameters = [], mapper } = query;
+    const compatibleQuery: WatchCompatibleQuery<RowType[]> = {
+      compile: () => ({
+        sql,
+        parameters
+      }),
+      execute: async ({ sql, parameters }) => {
+        const result = await this.getAll(sql, parameters);
+        return mapper ? result.map(mapper) : (result as RowType[]);
+      }
+    };
+    return this.customQuery(compatibleQuery);
+  }
+
+  /**
+   * Allows building a {@link WatchedQuery} using an existing {@link WatchCompatibleQuery}.
+   * The watched query will use the provided {@link WatchCompatibleQuery.execute} method to query results.
    *
-   * For a differential based watch , use {@link IncrementalWatchMode.DIFFERENTIAL}.
-   * See {@link DifferentialWatchedQueryBuilder} for more details.
    * @example
    * ```javascript
-   * const watchedQuery = powerSync
-   *  .incrementalWatch({ mode: IncrementalWatchMode.DIFFERENTIAL })
-   *  .build({
-   *    // ... Options
-   *  })
+   *
+   * // Potentially a query from an ORM like Drizzle
+   * const query = db.select().from(lists);
+   *
+   * const watchedTodos = powersync.customQuery(query)
+   * .watch()
+   * // OR use .differentialWatch() for fine-grained watches.
    * ```
    */
-  incrementalWatch<Mode extends IncrementalWatchMode>(options: { mode: Mode }): WatchedQueryBuilderMap[Mode] {
-    const { mode } = options;
-    const builderFactory = WatchedQueryBuilderMap[mode];
-    if (!builderFactory) {
-      throw new Error(
-        `Unsupported watch mode: ${mode}. Please specify one of [${Object.values(IncrementalWatchMode).join(', ')}]`
-      );
-    }
-    return builderFactory(this) as WatchedQueryBuilderMap[Mode];
+  customQuery<RowType>(query: WatchCompatibleQuery<RowType[]>): Query<RowType> {
+    return new CustomQuery({
+      db: this,
+      query
+    });
   }
 
   /**
@@ -944,21 +962,22 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     if (!onResult) {
       throw new Error('onResult is required');
     }
+    const { comparator } = options ?? {};
 
-    const { comparator = FalsyComparator } = 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<QueryResult | null>({
+    // This API yields a QueryResult type.
+    // This is not a standard Array result, which makes it incompatible with the .query API.
+    const watchedQuery = new OnChangeQueryProcessor({
+      db: this,
       comparator,
-      watch: {
+      placeholderData: null,
+      watchOptions: {
         query: {
           compile: () => ({
             sql: sql,
             parameters: parameters ?? []
           }),
           execute: () => this.executeReadOnly(sql, parameters)
         },
-        placeholderData: null,
         reportFetching: false,
         throttleMs: options?.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS
       }

packages/common/src/client/CustomQuery.ts

@@ -0,0 +1,55 @@
+import { AbstractPowerSyncDatabase } from './AbstractPowerSyncDatabase.js';
+import { Query, StandardWatchedQueryOptions } from './Query.js';
+import { FalsyComparator } from './watched/processors/comparators.js';
+import {
+  DifferentialQueryProcessor,
+  DifferentialWatchedQueryOptions
+} from './watched/processors/DifferentialQueryProcessor.js';
+import { OnChangeQueryProcessor } from './watched/processors/OnChangeQueryProcessor.js';
+import { DEFAULT_WATCH_QUERY_OPTIONS, WatchCompatibleQuery, WatchedQueryOptions } from './watched/WatchedQuery.js';
+
+/**
+ * @internal
+ */
+export interface CustomQueryOptions<RowType> {
+  db: AbstractPowerSyncDatabase;
+  query: WatchCompatibleQuery<RowType[]>;
+}
+
+/**
+ * @internal
+ */
+export class CustomQuery<RowType> implements Query<RowType> {
+  constructor(protected options: CustomQueryOptions<RowType>) {}
+
+  protected resolveOptions(options: WatchedQueryOptions) {
+    return {
+      reportFetching: options?.reportFetching ?? DEFAULT_WATCH_QUERY_OPTIONS.reportFetching,
+      throttleMs: options?.throttleMs ?? DEFAULT_WATCH_QUERY_OPTIONS.throttleMs
+    };
+  }
+
+  watch(watchOptions: StandardWatchedQueryOptions<RowType>) {
+    return new OnChangeQueryProcessor<RowType[]>({
+      db: this.options.db,
+      comparator: watchOptions?.comparator ?? FalsyComparator,
+      placeholderData: watchOptions?.placeholderData ?? [],
+      watchOptions: {
+        ...this.resolveOptions(watchOptions),
+        query: this.options.query
+      }
+    });
+  }
+
+  differentialWatch(differentialWatchOptions: DifferentialWatchedQueryOptions<RowType>) {
+    return new DifferentialQueryProcessor<RowType>({
+      db: this.options.db,
+      differentiator: differentialWatchOptions?.differentiator,
+      placeholderData: differentialWatchOptions?.placeholderData ?? [],
+      watchOptions: {
+        ...this.resolveOptions(differentialWatchOptions),
+        query: this.options.query
+      }
+    });
+  }
+}

packages/common/src/client/Query.ts

@@ -0,0 +1,88 @@
+import { ArrayComparator } from './watched/processors/comparators.js';
+import {
+  DifferentialWatchedQuery,
+  DifferentialWatchedQueryOptions
+} from './watched/processors/DifferentialQueryProcessor.js';
+import { ComparisonWatchedQuery } from './watched/processors/OnChangeQueryProcessor.js';
+import { WatchedQueryOptions } from './watched/WatchedQuery.js';
+
+/**
+ * Query parameters for {@link ArrayQueryDefinition.parameters}
+ */
+export type QueryParam = string | number | boolean | null | undefined | bigint | Uint8Array;
+
+/**
+ * Options for building a query with {@link AbstractPowerSyncDatabase.query}.
+ * This query will be executed with {@link AbstractPowerSyncDatabase.getAll}.
+ */
+export interface ArrayQueryDefinition<RowType = unknown> {
+  sql: string;
+  parameters?: ReadonlyArray<QueryParam>;
+  /**
+   * Maps the raw SQLite row to a custom typed object.
+   * @example
+   * ```javascript
+   * mapper: (row) => ({
+   *  ...row,
+   *  created_at: new Date(row.created_at as string),
+   * })
+   * ```
+   */
+  mapper?: (row: Record<string, unknown>) => RowType;
+}
+
+/**
+ * Options for {@link Query.watch}.
+ */
+export interface StandardWatchedQueryOptions<RowType> extends WatchedQueryOptions {
+  /**
+   * Optional comparator which processes the items of an array of rows.
+   * The comparator compares the result set rows by index using the {@link ArrayComparatorOptions.compareBy} function.
+   * The comparator reports a changed result set as soon as a row does not match the previous result set.
+   *
+   * @example
+   * ```javascript
+   * comparator: new ArrayComparator({
+   *     compareBy: (item) => JSON.stringify(item)
+   * })
+   * ```
+   */
+  comparator?: ArrayComparator<RowType>;
+
+  /**
+   * The initial data state reported while the query is loading for the first time.
+   * @default []
+   */
+  placeholderData?: RowType[];
+}
+
+export interface Query<RowType> {
+  /**
+   * Creates a {@link WatchedQuery} which watches and emits results of the linked query.
+   *
+   * By default the returned watched query will emit changes whenever a change to the underlying SQLite tables is made.
+   * These changes might not be relevant to the query, but the query will emit a new result set.
+   *
+   * A {@link StandardWatchedQueryOptions.comparator} can be provided to limit the data emissions. The watched query will still
+   * query the underlying DB on a underlying table changes, but the result will only be emitted if the comparator detects a change in the results.
+   *
+   * The comparator in this method is optimized and returns early as soon as it detects a change. Each data emission will correlate to a change in the result set,
+   * but note that the result set will not maintain internal object references to the previous result set. If internal object references are needed,
+   * consider using {@link Query.differentialWatch} instead.
+   */
+  watch(options?: StandardWatchedQueryOptions<RowType>): ComparisonWatchedQuery<RowType[]>;
+
+  /**
+   * Creates a {@link WatchedQuery} which watches and emits results of the linked query.
+   *
+   * This query method watches for changes in the underlying SQLite tables and runs the query on each table change.
+   * The difference between the current and previous result set is computed.
+   * The watched query will not emit changes if the result set is identical to the previous result set.
+   * If the result set is different, the watched query will emit the new result set and provide a detailed diff of the changes.
+   *
+   * The deep differentiation allows maintaining result set object references between result emissions.
+   * The {@link DifferentialWatchedQuery.state.data} array will contain the previous row references for unchanged rows.
+   * A detailed diff of the changes can be accessed via {@link DifferentialWatchedQuery.state.diff}.
+   */
+  differentialWatch(options?: DifferentialWatchedQueryOptions<RowType>): DifferentialWatchedQuery<RowType>;
+}