cleanup compatible queries

Author: stevensJourney

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

@@ -1,3 +1,4 @@
+import { CompiledQuery } from '../../types/types.js';
 import { BaseListener, BaseObserverInterface } from '../../utils/BaseObserver.js';
 
 export interface WatchedQueryState<Data> {
@@ -25,10 +26,16 @@ export interface WatchedQueryState<Data> {
   data: Data;
 }
 
+/**
+ * Similar to {@link CompiledQuery}, but used for watched queries.
+ * The parameters are not read-only, as they can be modified. This is useful for compatibility with
+ * PowerSync queries.
+ */
 export interface WatchCompiledQuery {
   sql: string;
   parameters: any[];
 }
+
 /**
  *
  * @internal
@@ -37,7 +44,7 @@ export interface WatchCompiledQuery {
  */
 export interface WatchCompatibleQuery<ResultType> {
   execute(compiled: WatchCompiledQuery): Promise<ResultType>;
-  compile(): WatchCompiledQuery;
+  compile(): CompiledQuery;
 }
 
 /**

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

@@ -80,7 +80,12 @@ export class OnChangeQueryProcessor<Data> extends AbstractQueryProcessor<Data> {
             const partialStateUpdate: Partial<WatchedQueryState<Data>> = {};
 
             // Always run the query if an underlaying table has changed
-            const result = await watchOptions.query.execute(compiledQuery);
+            const result = await watchOptions.query.execute({
+              sql: compiledQuery.sql,
+              // Allows casting from ReadOnlyArray[unknown] to Array<unknown>
+              // This allows simpler compatibility with PowerSync queries
+              parameters: [...compiledQuery.parameters]
+            });
 
             if (this.reportFetching) {
               partialStateUpdate.isFetching = false;

packages/kysely-driver/tests/sqlite/watch.test.ts

@@ -261,4 +261,38 @@ describe('Watch Tests', () => {
     expect(receivedWithManagedOverflowCount).greaterThan(2);
     expect(receivedWithManagedOverflowCount).toBeLessThanOrEqual(4);
   });
+
+  it('incremental watch should accept queries', async () => {
+    const query = db.selectFrom('assets').select(db.fn.count('assets.id').as('count'));
+
+    const watch = powerSyncDb.incrementalWatch({
+      watch: {
+        query,
+        placeholderData: []
+      }
+    });
+
+    const latestDataPromise = new Promise<Awaited<ReturnType<typeof query.execute>>>((resolve) => {
+      const dispose = watch.subscribe({
+        onData: (data) => {
+          if (data.length > 0) {
+            resolve(data);
+            dispose();
+          }
+        }
+      });
+    });
+
+    await db
+      .insertInto('assets')
+      .values({
+        id: sql`uuid()`,
+        make: 'test',
+        customer_id: sql`uuid()`
+      })
+      .execute();
+
+    const data = await latestDataPromise;
+    expect(data.length).equals(1);
+  });
 });

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

@@ -64,7 +64,11 @@ export const useSingleSuspenseQuery = <T = any>(
       refresh: async (signal) => {
         try {
           console.log('calling refresh for single query', key);
-          const result = await parsedQuery.execute(parsedQuery.compile());
+          const compiledQuery = parsedQuery.compile();
+          const result = await parsedQuery.execute({
+            sql: compiledQuery.sql,
+            parameters: [...compiledQuery.parameters]
+          });
           if (signal.aborted) {
             return; // Abort if the signal is already aborted
           }

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

@@ -16,7 +16,11 @@ export const useSingleQuery = <RowType = any>(options: InternalHookOptions<RowTy
     async (signal?: AbortSignal) => {
       setOutputState((prev) => ({ ...prev, isLoading: true, isFetching: true, error: undefined }));
       try {
-        const result = await query.execute(query.compile());
+        const compiledQuery = query.compile();
+        const result = await query.execute({
+          sql: compiledQuery.sql,
+          parameters: [...compiledQuery.parameters]
+        });
         if (signal.aborted) {
           return;
         }

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

@@ -1,9 +1,4 @@
-import {
-  AbstractPowerSyncDatabase,
-  CompilableQuery,
-  WatchCompatibleQuery,
-  WatchCompiledQuery
-} from '@powersync/common';
+import { AbstractPowerSyncDatabase, CompilableQuery, CompiledQuery, WatchCompatibleQuery } from '@powersync/common';
 import React from 'react';
 import { usePowerSync } from '../PowerSyncContext';
 import { AdditionalOptions } from './watch-types';
@@ -15,7 +10,7 @@ export type InternalHookOptions<DataType> = {
 };
 
 export const checkQueryChanged = <T>(query: WatchCompatibleQuery<T>, options: AdditionalOptions) => {
-  let _compiled: WatchCompiledQuery;
+  let _compiled: CompiledQuery;
   try {
     _compiled = query.compile();
   } catch (error) {