fix: diagnostics app dynamic schema (#737)

Author: stevensJourney

.changeset/loud-melons-try.md

@@ -0,0 +1,6 @@
+---
+'@powersync/diagnostics-app': patch
+---
+
+- 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,9 +4,10 @@ import {
   BucketChecksum,
   Checkpoint,
   ColumnType,
-  DBAdapter,
   isStreamingSyncCheckpoint,
+  isStreamingSyncCheckpointComplete,
   isStreamingSyncCheckpointDiff,
+  isStreamingSyncCheckpointPartiallyComplete,
   isStreamingSyncData,
   PowerSyncControlCommand,
   SqliteBucketStorage,
@@ -23,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;
   }
 
@@ -102,6 +103,12 @@ export class RustClientInterceptor extends SqliteBucketStorage {
       });
 
       await this.schemaManager.updateFromOperations(batch);
+    } else if (isStreamingSyncCheckpointPartiallyComplete(line) || isStreamingSyncCheckpointComplete(line)) {
+      // Refresh schema asynchronously, to allow us to better measure
+      // performance of initial sync.
+      setTimeout(() => {
+        this.schemaManager.refreshSchema(this.rdb);
+      }, 60);
     }
   }