Update generic typing and APIs

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -19,7 +19,6 @@ import { throttleTrailing } from '../utils/async.js';
 import { mutexRunExclusive } from '../utils/mutex.js';
 import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js';
-import { runOnSchemaChange } from './runOnSchemaChange.js';
 import { BucketStorageAdapter, PSInternalTable } from './sync/bucket/BucketStorageAdapter.js';
 import { CrudBatch } from './sync/bucket/CrudBatch.js';
 import { CrudEntry, CrudEntryJSON } from './sync/bucket/CrudEntry.js';
@@ -34,8 +33,7 @@ import {
   type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
 import { WatchedQuery } from './watched/WatchedQuery.js';
-import { WatchedQueryImpl } from './watched/WatchedQueryImpl.js';
-import { OnChangeQueryProcessor } from './watched/processors/OnChangeQueryProcessor.js';
+import { OnChangeQueryProcessor, WatchedQueryComparator } from './watched/processors/OnChangeQueryProcessor.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -89,6 +87,12 @@ export interface SQLWatchOptions {
    * Emits an empty result set immediately
    */
   triggerImmediate?: boolean;
+
+  /**
+   * 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>;
 }
 
 export interface WatchOnChangeEvent {
@@ -868,25 +872,28 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   }
 
   // TODO names
-  incrementalWatch<T>(options: {
+  incrementalWatch<DataType>(options: {
     sql: string;
     parameters?: any[];
     throttleMs?: number;
-    queryExecutor?: () => Promise<T[]>;
+    customExecutor?: {
+      initialData: DataType;
+      execute: () => Promise<DataType>;
+    };
     reportFetching?: boolean;
-  }): WatchedQuery<T> {
-    return new WatchedQueryImpl({
-      processor: new OnChangeQueryProcessor({
-        db: this,
-        compareBy: (item) => JSON.stringify(item), // TODO make configurable
-        watchedQuery: {
-          query: options.sql,
-          parameters: options.parameters,
-          throttleMs: options.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS,
-          queryExecutor: options.queryExecutor,
-          reportFetching: options.reportFetching
-        }
-      })
+  }): WatchedQuery<DataType> {
+    return new OnChangeQueryProcessor({
+      db: this,
+      comparator: {
+        checkEquality: (a, b) => JSON.stringify(a) == JSON.stringify(b)
+      },
+      query: {
+        sql: options.sql,
+        parameters: options.parameters,
+        throttleMs: options.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS,
+        customExecutor: options.customExecutor,
+        reportFetching: options.reportFetching
+      }
     });
   }
 
@@ -908,38 +915,42 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('onResult is required');
     }
 
-    const watchQuery = async (abortSignal: AbortSignal) => {
-      try {
-        const resolvedTables = await this.resolveTables(sql, parameters, options);
-        // Fetch initial data
-        const result = await this.executeReadOnly(sql, parameters);
-        onResult(result);
-
-        this.onChangeWithCallback(
-          {
-            onChange: async () => {
-              try {
-                const result = await this.executeReadOnly(sql, parameters);
-                onResult(result);
-              } catch (error) {
-                onError?.(error);
-              }
-            },
-            onError
+    const watch = new OnChangeQueryProcessor({
+      db: this,
+      // Comparisons are disabled if no comparator is provided
+      comparator: options?.comparator,
+      query: {
+        sql,
+        parameters,
+        throttleMs: options?.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS,
+        reportFetching: false,
+        // The default watch implementation returns QueryResult as the Data type
+        customExecutor: {
+          execute: async () => {
+            return this.executeReadOnly(sql, parameters);
           },
-          {
-            ...(options ?? {}),
-            tables: resolvedTables,
-            // Override the abort signal since we intercept it
-            signal: abortSignal
-          }
-        );
-      } catch (error) {
-        onError?.(error);
+          initialData: null
+        }
       }
-    };
+    });
 
-    runOnSchemaChange(watchQuery, this, options);
+    const dispose = watch.subscribe({
+      onData: (data) => {
+        if (!data) {
+          // This should not happen. We only use null for the initial data.
+          return;
+        }
+        onResult(data);
+      },
+      onError: (error) => {
+        onError(error);
+      }
+    });
+
+    options?.signal?.addEventListener('abort', () => {
+      dispose();
+      watch.close();
+    });
   }
 
   /**

packages/common/src/client/watched/WatchedQuery.ts

@@ -1,34 +1,43 @@
-import { DataStream } from '../../utils/DataStream.js';
-
-export interface WatchedQueryState<T> {
+export interface WatchedQueryState<Data> {
+  /**
+   * Indicates the initial loading state (hard loading). Loading becomes false once the first set of results from the watched query is available or an error occurs.
+   */
   isLoading: boolean;
+  /**
+   * Indicates whether the query is currently fetching data, is true during the initial load and any time when the query is re-evaluating (useful for large queries).
+   */
   isFetching: boolean;
+  /**
+   * The last error that occurred while executing the query.
+   */
   error: Error | null;
+  /**
+   * The last time the query was updated.
+   */
   lastUpdated: Date | null;
-  data: T[];
-}
-
-/**
- * Performs underlaying watching and yields a stream of results.
- * @internal
- */
-export interface WatchedQueryProcessor<T> {
-  readonly state: WatchedQueryState<T>;
-
-  generateStream(): Promise<DataStream<WatchedQueryState<T>>>;
-
-  updateQuery(query: WatchedQueryOptions<T>): void;
+  /**
+   * The last data returned by the query.
+   */
+  data: Data;
 }
 
 /**
  * @internal
  */
-export interface WatchedQueryOptions<T> {
-  query: string;
+export interface WatchedQueryOptions<DataType> {
+  sql: string;
   parameters?: any[];
   /** The minimum interval between queries. */
   throttleMs?: number;
-  queryExecutor?: () => Promise<T[]>;
+  /**
+   * Optional query executor responsible for executing the query.
+   * This can be used to return query results which are mapped from the database.
+   * Often this is useful for ORM queries or other query builders.
+   */
+  customExecutor?: {
+    execute: () => Promise<DataType>;
+    initialData: DataType;
+  };
   /**
    * If true (default) the watched query will update its state to report
    * on the fetching state of the query.
@@ -38,9 +47,32 @@ export interface WatchedQueryOptions<T> {
   reportFetching?: boolean;
 }
 
-export interface WatchedQuery<T> {
-  readonly state: WatchedQueryState<T>;
-  stream(): DataStream<WatchedQueryState<T>>;
-  updateQuery(query: WatchedQueryOptions<T>): void;
-  close(): void;
+export interface WatchedQuerySubscription<Data> {
+  onData?: (data: Data) => void | Promise<void>;
+  onError?: (error: Error) => void | Promise<void>;
+  onStateChange?: (state: WatchedQueryState<Data>) => void | Promise<void>;
+}
+
+export interface WatchedQuery<Data> {
+  /**
+   * Current state of the watched query.
+   */
+  readonly state: WatchedQueryState<Data>;
+
+  /**
+   * Subscribe to watched query events.
+   * @returns A function to unsubscribe from the events.
+   */
+  subscribe(subscription: WatchedQuerySubscription<Data>): () => void;
+
+  /**
+   * Updates the underlaying query.
+   * This will trigger a re-evaluation of the query and update the state.
+   */
+  updateQuery(query: WatchedQueryOptions<Data>): Promise<void>;
+
+  /**
+   * Close the watched query and end all subscriptions.
+   */
+  close(): Promise<void>;
 }