wip: query interface

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -32,7 +32,7 @@ import {
   type PowerSyncConnectionOptions,
   type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
-import { WatchedQuery } from './watched/WatchedQuery.js';
+import { WatchedQuery, WatchedQueryOptions } from './watched/WatchedQuery.js';
 import { OnChangeQueryProcessor, WatchedQueryComparator } from './watched/processors/OnChangeQueryProcessor.js';
 
 export interface DisconnectAndClearOptions {
@@ -871,30 +871,25 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     return this.watchWithAsyncGenerator(sql, parameters, options);
   }
 
-  // TODO names
-  incrementalWatch<DataType>(options: {
-    sql: string;
-    parameters?: any[];
-    throttleMs?: number;
-    customExecutor?: {
-      initialData: DataType;
-      execute: () => Promise<DataType>;
-    };
-    reportFetching?: boolean;
-  }): 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
-      }
-    });
+  // TODO names and types
+  incrementalWatch<DataType>(
+    options: { watchOptions: WatchedQueryOptions<DataType> } & {
+      mode: 'comparison';
+      comparator?: WatchedQueryComparator<DataType>;
+    }
+  ): WatchedQuery<DataType> {
+    switch (options.mode) {
+      case 'comparison':
+        return new OnChangeQueryProcessor({
+          db: this,
+          comparator: options.comparator ?? {
+            checkEquality: (a, b) => JSON.stringify(a) == JSON.stringify(b)
+          },
+          watchOptions: options.watchOptions
+        });
+      default:
+        throw new Error(`Invalid mode specified ${options.mode}`);
+    }
   }
 
   /**
@@ -919,18 +914,17 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       db: this,
       // Comparisons are disabled if no comparator is provided
       comparator: options?.comparator,
-      query: {
-        sql,
-        parameters,
+      watchOptions: {
+        placeholderData: null,
+        query: {
+          compile: () => ({
+            sql: sql,
+            parameters: parameters ?? []
+          }),
+          execute: () => this.executeReadOnly(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);
-          },
-          initialData: null
-        }
+        reportFetching: false
       }
     });
 

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

@@ -1,12 +1,15 @@
+import { CompiledQuery } from 'src/types/types.js';
 import { BaseListener, BaseObserverInterface } from '../../utils/BaseObserver.js';
 
 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.
+   * 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).
+   * 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;
   /**
@@ -23,23 +26,29 @@ export interface WatchedQueryState<Data> {
   data: Data;
 }
 
+/**
+ * @internal
+ * Similar to {@link CompatibleQuery}, except the `execute` method
+ * does not enforce an Array result type.
+ */
+export interface WatchCompatibleQuery<ResultType> {
+  execute(compiled: CompiledQuery): Promise<ResultType>;
+  compile(): CompiledQuery;
+}
+
 /**
  * @internal
  */
 export interface WatchedQueryOptions<DataType> {
-  sql: string;
-  parameters?: any[];
-  /** The minimum interval between queries. */
-  throttleMs?: number;
+  query: WatchCompatibleQuery<DataType>;
+
   /**
-   * 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.
+   * Initial result data which is presented while the initial loading is executing
    */
-  customExecutor?: {
-    execute: () => Promise<DataType>;
-    initialData: DataType;
-  };
+  placeholderData: DataType;
+
+  /** The minimum interval between queries. */
+  throttleMs?: number;
   /**
    * If true (default) the watched query will update its state to report
    * on the fetching state of the query.
@@ -87,10 +96,10 @@ export interface WatchedQuery<Data> extends BaseObserverInterface<WatchedQueryLi
   subscribe(subscription: WatchedQuerySubscription<Data>): () => void;
 
   /**
-   * Updates the underlaying query.
+   * Updates the underlaying query options.
    * This will trigger a re-evaluation of the query and update the state.
    */
-  updateQuery(query: WatchedQueryOptions<Data>): Promise<void>;
+  updateSettings(options: WatchedQueryOptions<Data>): Promise<void>;
 
   /**
    * Close the watched query and end all subscriptions.

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

@@ -15,7 +15,7 @@ import {
  */
 export interface AbstractQueryProcessorOptions<Data> {
   db: AbstractPowerSyncDatabase;
-  query: WatchedQueryOptions<Data>;
+  watchOptions: WatchedQueryOptions<Data>;
 }
 
 /**
@@ -64,22 +64,22 @@ export abstract class AbstractQueryProcessor<Data = unknown[]>
       isFetching: this.reportFetching, // Only set to true if we will report updates in future
       error: null,
       lastUpdated: null,
-      data: options.query.customExecutor?.initialData ?? ([] as Data)
+      data: options.watchOptions.placeholderData
     };
     this.initialized = this.init();
   }
 
   protected get reportFetching() {
-    return this.options.query.reportFetching ?? true;
+    return this.options.watchOptions.reportFetching ?? true;
   }
 
   /**
    * Updates the underlaying query.
    */
-  async updateQuery(query: WatchedQueryOptions<Data>) {
+  async updateSettings(query: WatchedQueryOptions<Data>) {
     await this.initialized;
 
-    this.options.query = query;
+    this.options.watchOptions = query;
     this.abortController.abort();
     this.abortController = new AbortController();
     await this.linkQuery({
@@ -116,7 +116,7 @@ export abstract class AbstractQueryProcessor<Data = unknown[]>
     db.registerListener({
       schemaChanged: async () => {
         await this.runWithReporting(async () => {
-          await this.updateQuery(this.options.query);
+          await this.updateSettings(this.options.watchOptions);
         });
       },
       closing: () => {
@@ -126,7 +126,7 @@ export abstract class AbstractQueryProcessor<Data = unknown[]>
 
     // Initial setup
     await this.runWithReporting(async () => {
-      await this.updateQuery(this.options.query);
+      await this.updateSettings(this.options.watchOptions);
     });
   }
 

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

@@ -62,10 +62,11 @@ export class OnChangeQueryProcessor<Data> extends AbstractQueryProcessor<Data> {
   }
 
   protected async linkQuery(options: LinkQueryOptions<Data>): Promise<void> {
-    const { db, query } = this.options;
+    const { db, watchOptions } = this.options;
     const { abortSignal } = options;
 
-    const tables = await db.resolveTables(query.sql, query.parameters);
+    const compiledQuery = watchOptions.query.compile();
+    const tables = await db.resolveTables(compiledQuery.sql, compiledQuery.parameters as any[]);
 
     db.onChangeWithCallback(
       {
@@ -79,9 +80,7 @@ export class OnChangeQueryProcessor<Data> extends AbstractQueryProcessor<Data> {
             const partialStateUpdate: Partial<WatchedQueryState<Data>> = {};
 
             // Always run the query if an underlaying table has changed
-            const result = query.customExecutor
-              ? await query.customExecutor.execute()
-              : ((await db.getAll(query.sql, query.parameters)) as Data);
+            const result = await watchOptions.query.execute(compiledQuery);
 
             if (this.reportFetching) {
               partialStateUpdate.isFetching = false;
@@ -110,7 +109,7 @@ export class OnChangeQueryProcessor<Data> extends AbstractQueryProcessor<Data> {
       {
         signal: abortSignal,
         tables,
-        throttleMs: query.throttleMs,
+        throttleMs: watchOptions.throttleMs,
         triggerImmediate: true // used to emit the initial state
       }
     );

packages/react/src/QueryStore.ts

@@ -1,33 +1,31 @@
-import { AbstractPowerSyncDatabase, WatchedQuery } from '@powersync/common';
-import { Query } from './WatchedQuery';
+import { AbstractPowerSyncDatabase, WatchCompatibleQuery, WatchedQuery } from '@powersync/common';
 import { AdditionalOptions } from './hooks/useQuery';
 
-export function generateQueryKey(sqlStatement: string, parameters: any[], options: AdditionalOptions): string {
+export function generateQueryKey(
+  sqlStatement: string,
+  parameters: ReadonlyArray<unknown>,
+  options: AdditionalOptions
+): string {
   return `${sqlStatement} -- ${JSON.stringify(parameters)} -- ${JSON.stringify(options)}`;
 }
 
 export class QueryStore {
-  cache = new Map<string, WatchedQuery<unknown[]>>();
+  cache = new Map<string, WatchedQuery<unknown>>();
 
   constructor(private db: AbstractPowerSyncDatabase) {}
 
-  getQuery(key: string, query: Query<unknown>, options: AdditionalOptions) {
+  getQuery(key: string, query: WatchCompatibleQuery<unknown>, options: AdditionalOptions) {
     if (this.cache.has(key)) {
       return this.cache.get(key);
     }
 
-    const customExecutor = typeof query.rawQuery !== 'string' ? query.rawQuery : null;
-
     const watchedQuery = this.db.incrementalWatch({
-      sql: query.sqlStatement,
-      parameters: query.queryParameters,
-      customExecutor: customExecutor
-        ? {
-            initialData: [],
-            execute: () => customExecutor.execute()
-          }
-        : undefined,
-      throttleMs: options.throttleMs
+      mode: 'comparison',
+      watchOptions: {
+        query,
+        placeholderData: [],
+        throttleMs: options.throttleMs
+      }
     });
 
     const disposer = watchedQuery.registerListener({

packages/react/src/WatchedQuery.ts

@@ -1,10 +1,8 @@
-import { CompilableQuery } from '@powersync/common';
-
-export class Query<T> {
-  rawQuery: string | CompilableQuery<T>;
-  sqlStatement: string;
-  queryParameters: any[];
-}
+// export class Query<T> {
+//   rawQuery: string | CompilableQuery<T>;
+//   sqlStatement: string;
+//   queryParameters: any[];
+// }
 
 // export interface WatchedQueryListener extends BaseListener {
 //onUpdate: () => void;