Make watched queries more robust for schema updating

Author: stevensJourney

.changeset/loud-melons-try.md

@@ -2,4 +2,5 @@
 '@powersync/diagnostics-app': patch
 ---
 
-Fixed bug where Rust client implementation would not update the dynamic schema after sync.
+- Fixed bug where Rust client implementation would not update the dynamic schema after sync.
+- Improved dynamic schema refresh logic for all implementations. Updating the schema should now always update all dependent watched queries e.g. in the SQL Console.

tools/diagnostics-app/src/library/powersync/ConnectionManager.ts

@@ -13,12 +13,12 @@ import {
   WebStreamingSyncImplementation,
   WebStreamingSyncImplementationOptions
 } from '@powersync/web';
+import React from 'react';
 import { safeParse } from '../safeParse/safeParse';
 import { DynamicSchemaManager } from './DynamicSchemaManager';
 import { RecordingStorageAdapter } from './RecordingStorageAdapter';
-import { TokenConnector } from './TokenConnector';
 import { RustClientInterceptor } from './RustClientInterceptor';
-import React from 'react';
+import { TokenConnector } from './TokenConnector';
 
 const baseLogger = createBaseLogger();
 baseLogger.useDefaults();
@@ -72,8 +72,8 @@ export async function connect() {
   const remote = new WebRemote(connector);
   const adapter =
     client == SyncClientImplementation.JAVASCRIPT
-      ? new RecordingStorageAdapter(db.database, schemaManager)
-      : new RustClientInterceptor(db.database, remote, schemaManager);
+      ? new RecordingStorageAdapter(db, schemaManager)
+      : new RustClientInterceptor(db, remote, schemaManager);
 
   const syncOptions: WebStreamingSyncImplementationOptions = {
     adapter,
@@ -99,7 +99,7 @@ export async function clearData() {
   await sync?.disconnect();
   await db.disconnectAndClear();
   await schemaManager.clear();
-  await schemaManager.refreshSchema(db.database);
+  await schemaManager.refreshSchema(db);
   if (connector.hasCredentials()) {
     const params = getParams();
     await sync?.connect({ params });

tools/diagnostics-app/src/library/powersync/DynamicSchemaManager.ts

@@ -1,4 +1,12 @@
-import { Column, ColumnType, DBAdapter, OpTypeEnum, Schema, SyncDataBatch, Table } from '@powersync/web';
+import {
+  AbstractPowerSyncDatabase,
+  Column,
+  ColumnType,
+  OpTypeEnum,
+  Schema,
+  SyncDataBatch,
+  Table
+} from '@powersync/web';
 import { AppSchema } from './AppSchema';
 import { JsSchemaGenerator } from './JsSchemaGenerator';
 
@@ -65,10 +73,10 @@ export class DynamicSchemaManager {
     }
   }
 
-  async refreshSchema(db: DBAdapter) {
+  async refreshSchema(db: AbstractPowerSyncDatabase) {
     if (this.dirty) {
-      const json = this.buildSchema().toJSON();
-      await db.execute('SELECT powersync_replace_schema(?)', [JSON.stringify(json)]);
+      // Use the PowerSyncDatabase since this will refresh all watched queries
+      await db.updateSchema(this.buildSchema());
       this.dirty = false;
       console.log('Updated dynamic schema:', this.tables);
     }

tools/diagnostics-app/src/library/powersync/RecordingStorageAdapter.ts

@@ -1,23 +1,16 @@
-import {
-  AbstractPowerSyncDatabase,
-  Checkpoint,
-  ColumnType,
-  DBAdapter,
-  SqliteBucketStorage,
-  SyncDataBatch
-} from '@powersync/web';
+import { AbstractPowerSyncDatabase, Checkpoint, ColumnType, SqliteBucketStorage, SyncDataBatch } from '@powersync/web';
 import { DynamicSchemaManager } from './DynamicSchemaManager';
 
 export class RecordingStorageAdapter extends SqliteBucketStorage {
-  private rdb: DBAdapter;
+  private rdb: AbstractPowerSyncDatabase;
 
   public tables: Record<string, Record<string, ColumnType>> = {};
 
   constructor(
-    db: DBAdapter,
+    db: AbstractPowerSyncDatabase,
     private schemaManager: DynamicSchemaManager
   ) {
-    super(db, (AbstractPowerSyncDatabase as any).transactionMutex);
+    super(db.database, (AbstractPowerSyncDatabase as any).transactionMutex);
     this.rdb = db;
   }
 

tools/diagnostics-app/src/library/powersync/RustClientInterceptor.ts

@@ -4,7 +4,6 @@ import {
   BucketChecksum,
   Checkpoint,
   ColumnType,
-  DBAdapter,
   isStreamingSyncCheckpoint,
   isStreamingSyncCheckpointComplete,
   isStreamingSyncCheckpointDiff,
@@ -25,17 +24,17 @@ import { DynamicSchemaManager } from './DynamicSchemaManager';
  * `powersync_control` calls to decode sync lines and derive progress information.
  */
 export class RustClientInterceptor extends SqliteBucketStorage {
-  private rdb: DBAdapter;
+  private rdb: AbstractPowerSyncDatabase;
   private lastStartedCheckpoint: Checkpoint | null = null;
 
   public tables: Record<string, Record<string, ColumnType>> = {};
 
   constructor(
-    db: DBAdapter,
+    db: AbstractPowerSyncDatabase,
     private remote: AbstractRemote,
     private schemaManager: DynamicSchemaManager
   ) {
-    super(db, (AbstractPowerSyncDatabase as any).transactionMutex);
+    super(db.database, (AbstractPowerSyncDatabase as any).transactionMutex);
     this.rdb = db;
   }