Using iterateAsyncListeners for schema change event.

Moved optionalSync test schema to separate file.

Author: Chriztiaan

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -103,7 +103,7 @@ export interface WatchOnChangeHandler {
 
 export interface PowerSyncDBListener extends StreamingSyncImplementationListener {
   initialized: () => void;
-  schemaChanged: (schema: Schema) => void;
+  schemaChanged: (schema: Schema) => Promise<void>;
 }
 
 export interface PowerSyncCloseOptions {
@@ -359,8 +359,9 @@ 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())]);
-    this.iterateListeners((cb) => cb.schemaChanged?.(schema));
+    await this.iterateAsyncListeners(async (cb) => cb.schemaChanged?.(schema));
   }
 
   /**
@@ -761,7 +762,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     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);
@@ -790,11 +790,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       }
     };
 
-    const triggerWatchedQuery = () => {
+    const triggerWatchedQuery = async () => {
       const abortController = new AbortController();
-
       let disposeSchemaListener: (() => void) | null = null;
-
       const stopWatching = () => {
         abortController.abort('Abort triggered');
         disposeSchemaListener?.();
@@ -804,16 +802,15 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       };
 
       options?.signal?.addEventListener('abort', stopWatching);
-
       disposeSchemaListener = this.registerListener({
-        schemaChanged: () => {
+        schemaChanged: async () => {
           stopWatching();
           // Re trigger the watched query (recursively)
-          triggerWatchedQuery();
+          await triggerWatchedQuery();
         }
       });
 
-      return watchQuery(abortController.signal);
+      await watchQuery(abortController.signal);
     };
 
     triggerWatchedQuery();

packages/web/tests/offline.test.ts

@@ -1,67 +1,12 @@
 import { AbstractPowerSyncDatabase, column, Schema, Table } from '@powersync/common';
 import { PowerSyncDatabase } from '@powersync/web';
 import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { makeOptionalSyncSchema } from './utils/optionalSyncTestSchema';
 
 const assetId = '2290de4f-0488-4e50-abed-f8e8eb1d0b42';
 const userId = '3390de4f-0488-4e50-abed-f8e8eb1d0b42';
 const customerId = '4490de4f-0488-4e50-abed-f8e8eb1d0b42';
 
-const assetsDef = {
-  name: 'assets',
-  columns: {
-    created_at: column.text,
-    make: column.text,
-    model: column.text,
-    serial_number: column.text,
-    quantity: column.integer,
-    user_id: column.text,
-    customer_id: column.text,
-    description: column.text
-  },
-  options: { indexes: { makemodel: ['make, model'] } }
-};
-
-const customersDef = {
-  name: 'customers',
-  columns: {
-    name: column.text,
-    email: column.text
-  },
-  options: {}
-};
-
-function makeSchema(synced: boolean) {
-  const syncedName = (table: string): string => {
-    if (synced) {
-      return table;
-    } else {
-      return `inactive_synced_${table}`;
-    }
-  };
-
-  const localName = (table: string): string => {
-    if (synced) {
-      return `inactive_local_${table}`;
-    } else {
-      return table;
-    }
-  };
-  return new Schema({
-    assets: new Table(assetsDef.columns, { ...assetsDef.options, viewName: syncedName(assetsDef.name) }),
-    local_assets: new Table(assetsDef.columns, {
-      ...assetsDef.options,
-      localOnly: true,
-      viewName: localName(assetsDef.name)
-    }),
-    customers: new Table(customersDef.columns, { ...customersDef.options, viewName: syncedName(customersDef.name) }),
-    local_customers: new Table(customersDef.columns, {
-      ...customersDef.options,
-      localOnly: true,
-      viewName: localName(customersDef.name)
-    })
-  });
-}
-
 describe('Schema Tests', () => {
   let db: AbstractPowerSyncDatabase;
 
@@ -73,7 +18,7 @@ describe('Schema Tests', () => {
        * consistent
        */
       database: { dbFilename: 'test.db' },
-      schema: makeSchema(false),
+      schema: makeOptionalSyncSchema(false),
       flags: {
         enableMultiTabs: false
       }
@@ -98,7 +43,7 @@ describe('Schema Tests', () => {
     expect(await db.getAll('SELECT data FROM ps_crud ORDER BY id')).toEqual([]);
 
     // Now switch to the "online" schema
-    await db.updateSchema(makeSchema(true));
+    await db.updateSchema(makeOptionalSyncSchema(true));
 
     // Note that updateSchema cannot be called inside a transaction, and there
     // is a possibility of crash between updating the schema, and when the data
@@ -148,7 +93,7 @@ describe('Schema Tests', () => {
     const customerWatchTables = await getSourceTables(db, 'SELECT * FROM customers');
     expect(customerWatchTables.includes('ps_data_local__local_customers')).toBeTruthy();
 
-    await db.updateSchema(makeSchema(true));
+    await db.updateSchema(makeOptionalSyncSchema(true));
 
     const onlineCustomerWatchTables = await getSourceTables(db, 'SELECT * FROM customers');
     expect(onlineCustomerWatchTables.includes('ps_data__customers')).toBeTruthy();

packages/web/tests/utils/optionalSyncTestSchema.ts

@@ -0,0 +1,58 @@
+import { column, Schema, Table } from '@powersync/common';
+
+const assetsDef = {
+  name: 'assets',
+  columns: {
+    created_at: column.text,
+    make: column.text,
+    model: column.text,
+    serial_number: column.text,
+    quantity: column.integer,
+    user_id: column.text,
+    customer_id: column.text,
+    description: column.text
+  },
+  options: {}
+};
+
+const customersDef = {
+  name: 'customers',
+  columns: {
+    name: column.text,
+    email: column.text
+  },
+  options: {}
+};
+
+export function makeOptionalSyncSchema(synced: boolean) {
+  const syncedName = (table: string): string => {
+    if (synced) {
+      return table;
+    } else {
+      return `inactive_synced_${table}`;
+    }
+  };
+
+  const localName = (table: string): string => {
+    if (synced) {
+      return `inactive_local_${table}`;
+    } else {
+      return table;
+    }
+  };
+
+  return new Schema({
+    assets: new Table(assetsDef.columns, { ...assetsDef.options, viewName: syncedName(assetsDef.name) }),
+    local_assets: new Table(assetsDef.columns, {
+      ...assetsDef.options,
+      localOnly: true,
+      viewName: localName(assetsDef.name)
+    }),
+    customers: new Table(customersDef.columns, { ...customersDef.options, viewName: syncedName(customersDef.name) }),
+    local_customers: new Table(customersDef.columns, {
+      ...customersDef.options,
+      localOnly: true,
+      viewName: localName(customersDef.name)
+    })
+  });
+}

packages/web/tests/watch.test.ts

@@ -1,8 +1,9 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { v4 as uuid } from 'uuid';
-import { AbstractPowerSyncDatabase } from '@powersync/common';
+import { AbstractPowerSyncDatabase, QueryResult } from '@powersync/common';
 import { PowerSyncDatabase } from '@powersync/web';
 import { testSchema } from './utils/testDb';
+import { makeOptionalSyncSchema } from './utils/optionalSyncTestSchema';
 vi.useRealTimers();
 
 /**
@@ -26,6 +27,7 @@ describe('Watch Tests', () => {
         enableMultiTabs: false
       }
     });
+    await powersync.init();
   });
 
   afterEach(async () => {
@@ -83,6 +85,7 @@ describe('Watch Tests', () => {
     const receivedUpdates = new Promise<void>((resolve) => {
       const onUpdate = () => {
         receivedUpdatesCount++;
+
         if (receivedUpdatesCount == updatesCount) {
           abortController.abort();
           resolve();
@@ -286,8 +289,6 @@ describe('Watch Tests', () => {
   });
 
   it('should throttle watch callback overflow', async () => {
-    const abortController = new AbortController();
-
     const updatesCount = 25;
 
     let receivedWithManagedOverflowCount = 0;
@@ -300,7 +301,7 @@ describe('Watch Tests', () => {
       'SELECT count() AS count FROM assets',
       [],
       { onResult: onResultOverflow },
-      { signal: overflowAbortController.signal, throttleMs: 1 }
+      { signal: overflowAbortController.signal, throttleMs: 100 }
     );
 
     // Allows us to count the number of updates received without the initial trigger
@@ -313,10 +314,9 @@ describe('Watch Tests', () => {
 
     await new Promise<void>((resolve) => setTimeout(resolve, 1 * throttleDuration));
 
-    abortController.abort();
     overflowAbortController.abort();
 
-    // Initial onResult plus two left after overflow was throttled for onChange triggers
-    expect(receivedWithManagedOverflowCount).toBe(3);
+    // Initial onResult plus 1 left after overflow was throttled for onChange triggers
+    expect(receivedWithManagedOverflowCount).toBe(2);
   });
 });

packages/web/tests/watchSchemaChange.test.ts

@@ -0,0 +1,85 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { v4 as uuid } from 'uuid';
+import { AbstractPowerSyncDatabase, QueryResult } from '@powersync/common';
+import { PowerSyncDatabase } from '@powersync/web';
+import { testSchema } from './utils/testDb';
+import { makeOptionalSyncSchema } from './utils/optionalSyncTestSchema';
+vi.useRealTimers();
+
+/**
+ * There seems to be an issue with Vitest browser mode's setTimeout and
+ * fake timer functionality.
+ * e.g. calling:
+ *      await new Promise<void>((resolve) => setTimeout(resolve, 10));
+ * waits for 1 second instead of 10ms.
+ * Setting this to 1 second as a work around.
+ */
+const throttleDuration = 1000;
+
+describe('Watch With Schema Change Tests', () => {
+  let powersync: AbstractPowerSyncDatabase;
+
+  beforeEach(async () => {
+    powersync = new PowerSyncDatabase({
+      database: { dbFilename: 'test-watch-optional-sync.db' },
+      schema: makeOptionalSyncSchema(false),
+      flags: {
+        enableMultiTabs: false
+      }
+    });
+    await powersync.init();
+  });
+
+  afterEach(async () => {
+    await powersync.disconnectAndClear();
+    await powersync.close();
+  });
+
+  it('should trigger onResult after schema change', async () => {
+    const userId = uuid();
+    const abortController = new AbortController();
+
+    let lastResult = 0;
+
+    const onUpdate = (results: QueryResult) => {
+      lastResult = results.rows?._array[0].count; // `count` from sql output
+    };
+
+    powersync.watch(
+      'SELECT count() AS count FROM assets',
+      [],
+      { onResult: onUpdate },
+      { signal: abortController.signal, throttleMs: throttleDuration }
+    );
+
+    // Insert 3 records
+    for (let i = 0; i < 3; i++) {
+      await powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['test', userId]);
+    }
+    await new Promise<void>((resolve) => setTimeout(resolve, throttleDuration * 2));
+
+    expect(lastResult).toBe(3);
+
+    await powersync.updateSchema(makeOptionalSyncSchema(true));
+
+    await powersync.writeTransaction(async (tx) => {
+      // Copy local data to the "online" views.
+
+      await tx.execute(
+        'INSERT INTO assets(id, description, customer_id, user_id) SELECT id, description, customer_id, ? FROM inactive_local_assets',
+        [userId]
+      );
+
+      // Delete the "local-only" data.
+      await tx.execute('DELETE FROM inactive_local_customers');
+      await tx.execute('DELETE FROM inactive_local_assets');
+    });
+
+    for (let i = 0; i < 3; i++) {
+      await powersync.execute('INSERT INTO assets(id, make, customer_id) VALUES (uuid(), ?, ?)', ['test', uuid()]);
+    }
+    await new Promise<void>((resolve) => setTimeout(resolve, throttleDuration * 2));
+
+    expect(lastResult).toBe(6);
+  });
+});