feat: refresh watched queries on schema changes (#377)

Co-authored-by: Steven Ontong <[email protected]>

Author: Chriztiaan

.changeset/itchy-years-drop.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': minor
+---
+
+Updated watch functions to recalculate depedendent tables if schema is updated.

.changeset/short-owls-play.md

@@ -0,0 +1,7 @@
+---
+'@powersync/op-sqlite': minor
+'@powersync/react-native': minor
+'@powersync/web': minor
+---
+
+Added `refreshSchema()` which will cause all connections to be aware of a schema change.

.changeset/ten-birds-camp.md

@@ -0,0 +1,5 @@
+---
+'@powersync/vue': patch
+---
+
+Queries will recalculate dependent tables if schema is updated.

.changeset/tender-llamas-shop.md

@@ -0,0 +1,6 @@
+---
+'@powersync/tanstack-react-query': patch
+'@powersync/react': patch
+---
+
+Queries will recalculate dependent tables if schema is updated.

demos/react-supabase-todolist-optional-sync/src/app/views/layout.tsx

@@ -58,7 +58,7 @@ export default function ViewsLayout({ children }: { children: React.ReactNode })
         beforeNavigate: async () => {
           // If user is logged in, sign out and stay on the current page
           if (supabase?.currentSession) {
-            await supabase?.client.auth.signOut();
+            await supabase?.logout();
             await powerSync.disconnectAndClear();
             setSyncEnabled(powerSync.database.name, false);
 

demos/react-supabase-todolist-optional-sync/src/library/powersync/SupabaseConnector.ts

@@ -93,6 +93,11 @@ export class SupabaseConnector extends BaseObserver<SupabaseConnectorListener> i
     this.updateSession(session);
   }
 
+  async logout() {
+    await this.client.auth.signOut();
+    this.updateSession(null);
+  }
+
   async fetchCredentials() {
     const {
       data: { session },

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -28,6 +28,7 @@ import {
   StreamingSyncImplementation,
   StreamingSyncImplementationListener
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
+import { runOnSchemaChange } from './runOnSchemaChange.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -103,6 +104,7 @@ export interface WatchOnChangeHandler {
 
 export interface PowerSyncDBListener extends StreamingSyncImplementationListener {
   initialized: () => void;
+  schemaChanged: (schema: Schema) => void;
 }
 
 export interface PowerSyncCloseOptions {
@@ -360,7 +362,10 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       this.options.logger?.warn('Schema validation failed. Unexpected behaviour could occur', ex);
     }
     this._schema = schema;
+
     await this.database.execute('SELECT powersync_replace_schema(?)', [JSON.stringify(this.schema.toJSON())]);
+    await this.database.refreshSchema();
+    this.iterateListeners(async (cb) => cb.schemaChanged?.(schema));
   }
 
   /**
@@ -758,10 +763,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('onResult is required');
     }
 
-    (async () => {
+    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);
@@ -780,13 +784,17 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
           },
           {
             ...(options ?? {}),
-            tables: resolvedTables
+            tables: resolvedTables,
+            // Override the abort signal since we intercept it
+            signal: abortSignal
           }
         );
       } catch (error) {
         onError?.(error);
       }
-    })();
+    };
+
+    runOnSchemaChange(watchQuery, this, options);
   }
 
   /**
@@ -796,19 +804,20 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    */
   watchWithAsyncGenerator(sql: string, parameters?: any[], options?: SQLWatchOptions): AsyncIterable<QueryResult> {
     return new EventIterator<QueryResult>((eventOptions) => {
-      (async () => {
-        const resolvedTables = await this.resolveTables(sql, parameters, options);
+      const handler: WatchHandler = {
+        onResult: (result) => {
+          eventOptions.push(result);
+        },
+        onError: (error) => {
+          eventOptions.fail(error);
+        }
+      };
 
-        // Fetch initial data
-        eventOptions.push(await this.executeReadOnly(sql, parameters));
+      this.watchWithCallback(sql, parameters, handler, options);
 
-        for await (const event of this.onChangeWithAsyncGenerator({
-          ...(options ?? {}),
-          tables: resolvedTables
-        })) {
-          eventOptions.push(await this.executeReadOnly(sql, parameters));
-        }
-      })();
+      options?.signal?.addEventListener('abort', () => {
+        eventOptions.stop();
+      });
     });
   }
 

packages/common/src/client/runOnSchemaChange.ts

@@ -0,0 +1,31 @@
+import { AbstractPowerSyncDatabase, SQLWatchOptions } from './AbstractPowerSyncDatabase.js';
+
+export function runOnSchemaChange(
+  callback: (signal: AbortSignal) => void,
+  db: AbstractPowerSyncDatabase,
+  options?: SQLWatchOptions
+): void {
+  const triggerWatchedQuery = () => {
+    const abortController = new AbortController();
+    let disposeSchemaListener: (() => void) | null = null;
+    const stopWatching = () => {
+      abortController.abort('Abort triggered');
+      disposeSchemaListener?.();
+      disposeSchemaListener = null;
+      // Stop listening to upstream abort for this watch
+      options?.signal?.removeEventListener('abort', stopWatching);
+    };
+
+    options?.signal?.addEventListener('abort', stopWatching);
+    disposeSchemaListener = db.registerListener({
+      schemaChanged: async () => {
+        stopWatching();
+        // Re trigger the watched query (recursively), setTimeout ensures that we don't modify the list of listeners while iterating through them
+        setTimeout(() => triggerWatchedQuery(), 0);
+      }
+    });
+    callback(abortController.signal);
+  };
+
+  triggerWatchedQuery();
+}

packages/common/src/db/DBAdapter.ts

@@ -101,6 +101,10 @@ export interface DBAdapter extends BaseObserverInterface<DBAdapterListener>, DBG
   readTransaction: <T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions) => Promise<T>;
   writeLock: <T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions) => Promise<T>;
   writeTransaction: <T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions) => Promise<T>;
+  /**
+   * This method refreshes the schema information across all connections. This is for advanced use cases, and should generally not be needed.
+   */ 
+  refreshSchema: () => Promise<void>;
 }
 
 export function isBatchedUpdateNotification(

packages/common/src/index.ts

@@ -4,6 +4,7 @@ export * from './client/SQLOpenFactory.js';
 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 { UpdateType, CrudEntry, OpId } from './client/sync/bucket/CrudEntry.js';
 export * from './client/sync/bucket/SqliteBucketStorage.js';
 export * from './client/sync/bucket/CrudBatch.js';

packages/powersync-op-sqlite/src/db/OPSQLiteConnection.ts

@@ -79,4 +79,8 @@ export class OPSQLiteConnection extends BaseObserver<DBAdapterListener> {
     }
     return result as T;
   }
+
+  async refreshSchema() {
+    await this.get("PRAGMA table_info('sqlite_master')");
+  }
 }

packages/powersync-op-sqlite/src/db/OPSqliteAdapter.ts

@@ -271,4 +271,13 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
       throw ex;
     }
   }
+
+  async refreshSchema(): Promise<void> {
+    await this.initialized;
+    await this.writeConnection!.refreshSchema();
+
+    for (let readConnection of this.readConnections) {
+      await readConnection.connection.refreshSchema();
+    }
+  }
 }

packages/react-native/package.json

@@ -30,8 +30,8 @@
   },
   "homepage": "https://docs.powersync.com/",
   "peerDependencies": {
-    "@journeyapps/react-native-quick-sqlite": "^2.0.0",
-    "@powersync/common": "workspace:^1.20.2",
+    "@journeyapps/react-native-quick-sqlite": "^2.1.0",
+    "@powersync/common": "workspace:^1.20.1",
     "react": "*",
     "react-native": "*"
   },
@@ -46,7 +46,7 @@
   },
   "devDependencies": {
     "@craftzdog/react-native-buffer": "^6.0.5",
-    "@journeyapps/react-native-quick-sqlite": "^2.0.0",
+    "@journeyapps/react-native-quick-sqlite": "^2.1.0",
     "@rollup/plugin-alias": "^5.1.0",
     "@rollup/plugin-commonjs": "^25.0.7",
     "@rollup/plugin-inject": "^5.0.5",

packages/react-native/src/db/adapters/react-native-quick-sqlite/RNQSDBAdapter.ts

@@ -124,4 +124,8 @@ export class RNQSDBAdapter extends BaseObserver<DBAdapterListener> implements DB
       }
     };
   }
+
+  async refreshSchema() {
+    await this.baseDB.refreshSchema();
+  }
 }

packages/react/src/WatchedQuery.ts

@@ -1,4 +1,11 @@
-import { AbstractPowerSyncDatabase, BaseListener, BaseObserver, CompilableQuery, Disposable } from '@powersync/common';
+import {
+  AbstractPowerSyncDatabase,
+  BaseListener,
+  BaseObserver,
+  CompilableQuery,
+  Disposable,
+  runOnSchemaChange
+} from '@powersync/common';
 import { AdditionalOptions } from './hooks/useQuery';
 
 export class Query<T> {
@@ -117,7 +124,7 @@ export class WatchedQuery extends BaseObserver<WatchedQueryListener> implements
       this.setError(error);
     };
 
-    (async () => {
+    const watchQuery = async (abortSignal: AbortSignal) => {
       await this.fetchTables();
       await this.fetchData();
 
@@ -131,12 +138,13 @@ export class WatchedQuery extends BaseObserver<WatchedQueryListener> implements
           },
           {
             ...this.options,
-            signal: this.controller.signal,
+            signal: abortSignal,
             tables: this.tables
           }
         );
       }
-    })();
+    };
+    runOnSchemaChange(watchQuery, this.db, { signal: this.controller.signal });
   }
 
   private setData(results: any[]) {

packages/react/src/hooks/useQuery.ts

@@ -118,10 +118,18 @@ export const useQuery = <T = any>(
   };
 
   React.useEffect(() => {
-    (async () => {
+    const updateData = async () => {
       await fetchTables();
       await fetchData();
-    })();
+    };
+
+    updateData();
+
+    const l = powerSync.registerListener({
+      schemaChanged: updateData
+    });
+
+    return () => l?.();
   }, [powerSync, memoizedParams, sqlStatement]);
 
   React.useEffect(() => {

packages/react/tests/useQuery.test.tsx

@@ -7,9 +7,7 @@ import { useQuery } from '../src/hooks/useQuery';
 
 const mockPowerSync = {
   currentStatus: { status: 'initial' },
-  registerListener: vi.fn(() => ({
-    statusChanged: vi.fn(() => 'updated')
-  })),
+  registerListener: vi.fn(() => {}),
   resolveTables: vi.fn(() => ['table1', 'table2']),
   onChangeWithCallback: vi.fn(),
   getAll: vi.fn(() => Promise.resolve(['list1', 'list2']))
@@ -92,9 +90,7 @@ describe('useQuery', () => {
   it('should set error when error occurs and runQueryOnce flag is set', async () => {
     const mockPowerSyncError = {
       currentStatus: { status: 'initial' },
-      registerListener: vi.fn(() => ({
-        statusChanged: vi.fn(() => 'updated')
-      })),
+      registerListener: vi.fn(() => {}),
       onChangeWithCallback: vi.fn(),
       resolveTables: vi.fn(() => ['table1', 'table2']),
       getAll: vi.fn(() => {
@@ -119,9 +115,7 @@ describe('useQuery', () => {
   it('should set error when error occurs', async () => {
     const mockPowerSyncError = {
       currentStatus: { status: 'initial' },
-      registerListener: vi.fn(() => ({
-        statusChanged: vi.fn(() => 'updated')
-      })),
+      registerListener: vi.fn(() => {}),
       onChangeWithCallback: vi.fn(),
       resolveTables: vi.fn(() => ['table1', 'table2']),
       getAll: vi.fn(() => {

packages/react/tests/useStatus.test.tsx

@@ -33,9 +33,7 @@ describe('useStatus', () => {
   it.skip('should update the status when the listener is called', () => {
     const mockPowerSyncInTest = {
       currentStatus: { status: 'initial' },
-      registerListener: () => ({
-        statusChanged: () => 'updated'
-      })
+      registerListener: vi.fn(() => {})
     };
 
     const wrapper = ({ children }) => (

packages/react/tests/useSuspenseQuery.test.tsx

@@ -11,9 +11,7 @@ const defaultQueryResult = ['list1', 'list2'];
 const createMockPowerSync = () => {
   return {
     currentStatus: { status: 'initial' },
-    registerListener: vi.fn(() => ({
-      statusChanged: vi.fn(() => 'updated')
-    })),
+    registerListener: vi.fn(() => {}),
     resolveTables: vi.fn(() => ['table1', 'table2']),
     onChangeWithCallback: vi.fn(),
     getAll: vi.fn(() => Promise.resolve(defaultQueryResult)) as Mock<any, any>

packages/tanstack-react-query/src/hooks/useQuery.ts

@@ -102,11 +102,20 @@ function useQueryCore<
   };
 
   React.useEffect(() => {
-    if (!query) return;
+    if (!query) return () => {};
 
     (async () => {
       await fetchTables();
     })();
+
+    const l = powerSync.registerListener({
+      schemaChanged: async () => {
+        await fetchTables();
+        queryClient.invalidateQueries({ queryKey: options.queryKey });
+      }
+    });
+
+    return () => l?.();
   }, [powerSync, sqlStatement, stringifiedParams]);
 
   const queryFn = React.useCallback(async () => {