Introducing WIP compilableQueryWatch().

Author: Chriztiaan

packages/common/src/client/compilableQueryWatch.ts

@@ -0,0 +1,55 @@
+import { CompilableQuery } from './../types/types.js';
+import { AbstractPowerSyncDatabase, SQLWatchOptions } from './AbstractPowerSyncDatabase.js';
+import { runOnSchemaChange } from './runOnSchemaChange.js';
+
+export interface CompilableQueryWatchHandler<T> {
+  onResult: (results: T[]) => void;
+  onError?: (error: Error) => void;
+}
+
+export function compilableQueryWatch<T>(
+  db: AbstractPowerSyncDatabase,
+  query: CompilableQuery<T>,
+  handler: CompilableQueryWatchHandler<T>,
+  options?: SQLWatchOptions
+): void {
+  const { onResult, onError = (e: Error) => {} } = handler ?? {};
+  if (!onResult) {
+    throw new Error('onResult is required');
+  }
+
+  const watchQuery = async (abortSignal: AbortSignal) => {
+    try {
+      const toSql = query.compile();
+      const resolvedTables = await db.resolveTables(toSql.sql, toSql.parameters as [], options);
+
+      // Fetch initial data
+      const result = await query.execute();
+      onResult(result);
+
+      db.onChangeWithCallback(
+        {
+          onChange: async () => {
+            try {
+              const result = await query.execute();
+              onResult(result);
+            } catch (error: any) {
+              onError(error);
+            }
+          },
+          onError
+        },
+        {
+          ...(options ?? {}),
+          tables: resolvedTables,
+          // Override the abort signal since we intercept it
+          signal: abortSignal
+        }
+      );
+    } catch (error: any) {
+      onError(error);
+    }
+  };
+
+  runOnSchemaChange(watchQuery, db, options);
+}

packages/common/src/index.ts

@@ -5,6 +5,7 @@ export * from './client/connection/PowerSyncBackendConnector.js';
 export * from './client/connection/PowerSyncCredentials.js';
 export * from './client/sync/bucket/BucketStorageAdapter.js';
 export { runOnSchemaChange } from './client/runOnSchemaChange.js';
+export { CompilableQueryWatchHandler, compilableQueryWatch } from './client/compilableQueryWatch.js';
 export { UpdateType, CrudEntry, OpId } from './client/sync/bucket/CrudEntry.js';
 export * from './client/sync/bucket/SqliteBucketStorage.js';
 export * from './client/sync/bucket/CrudBatch.js';

packages/drizzle-driver/src/sqlite/db.ts

@@ -1,5 +1,7 @@
 import {
   AbstractPowerSyncDatabase,
+  compilableQueryWatch,
+  CompilableQueryWatchHandler,
   QueryResult,
   runOnSchemaChange,
   SQLWatchOptions,
@@ -19,8 +21,9 @@ import { BaseSQLiteDatabase } from 'drizzle-orm/sqlite-core/db';
 import { SQLiteAsyncDialect } from 'drizzle-orm/sqlite-core/dialect';
 import type { DrizzleConfig } from 'drizzle-orm/utils';
 import { PowerSyncSQLiteSession, PowerSyncSQLiteTransactionConfig } from './sqlite-session';
+import { toCompilableQuery } from 'src/utils/compilableQuery';
 
-type WatchQuery = { toSQL(): Query; execute(): Promise<any> };
+type WatchQuery<T> = { toSQL(): Query; execute(): Promise<T> };
 
 export interface PowerSyncSQLiteDatabase<TSchema extends Record<string, unknown> = Record<string, never>>
   extends BaseSQLiteDatabase<'async', QueryResult, TSchema> {
@@ -31,7 +34,7 @@ export interface PowerSyncSQLiteDatabase<TSchema extends Record<string, unknown>
     config?: PowerSyncSQLiteTransactionConfig
   ): Promise<T>;
 
-  watch(query: WatchQuery, handler?: WatchHandler, options?: SQLWatchOptions): void;
+  watch<T>(query: WatchQuery<T>, handler?: CompilableQueryWatchHandler<T>, options?: SQLWatchOptions): void;
 }
 
 export function wrapPowerSyncWithDrizzle<TSchema extends Record<string, unknown> = Record<string, never>>(
@@ -60,50 +63,17 @@ export function wrapPowerSyncWithDrizzle<TSchema extends Record<string, unknown>
     logger
   });
 
-  const watch = (query: WatchQuery, handler?: WatchHandler, options?: SQLWatchOptions): void => {
-    const { onResult, onError = (e: Error) => {} } = handler ?? {};
-    if (!onResult) {
-      throw new Error('onResult is required');
-    }
-
-    const watchQuery = async (abortSignal: AbortSignal) => {
-      try {
-        const toSql = query.toSQL();
-        const resolvedTables = await db.resolveTables(toSql.sql, toSql.params, options);
-
-        // Fetch initial data
-        const result = await query.execute();
-        onResult(result);
-
-        db.onChangeWithCallback(
-          {
-            onChange: async () => {
-              try {
-                const result = await query.execute();
-                onResult(result);
-              } catch (error: any) {
-                onError(error);
-              }
-            },
-            onError
-          },
-          {
-            ...(options ?? {}),
-            tables: resolvedTables,
-            // Override the abort signal since we intercept it
-            signal: abortSignal
-          }
-        );
-      } catch (error: any) {
-        onError(error);
-      }
-    };
-
-    runOnSchemaChange(watchQuery, db, options);
-  };
+  // const watch = <T>(
+  //   query: WatchQuery<T>,
+  //   handler?: CompilableQueryWatchHandler<T>,
+  //   options?: SQLWatchOptions
+  // ): void => {
+  //   compilableQueryWatch(db, toCompilableQuery<T>(query), handler, options);
+  // };
 
   const baseDatabase = new BaseSQLiteDatabase('async', dialect, session, schema) as PowerSyncSQLiteDatabase<TSchema>;
   return Object.assign(baseDatabase, {
-    watch: (query: WatchQuery, handler?: WatchHandler, options?: SQLWatchOptions) => watch(query, handler, options)
+    watch: <T>(query: WatchQuery<T>, handler?: CompilableQueryWatchHandler<T>, options?: SQLWatchOptions) => {}
+    // watch(query, handler, options)
   });
 }

packages/drizzle-driver/src/utils/compilableQuery.ts

@@ -20,10 +20,7 @@ import { Query } from 'drizzle-orm';
  *   </View>
  * );
  */
-export function toCompilableQuery<T>(query: {
-  execute: () => Promise<T | T[]>;
-  toSQL: () => Query;
-}): CompilableQuery<T> {
+export function toCompilableQuery<T>(query: { execute: () => Promise<T[]>; toSQL: () => Query }): CompilableQuery<T> {
   return {
     compile: () => {
       const sql = query.toSQL();