[Feature] Table View Overrides (#45)

* flush table change updates on trailing edge. Prevents race conditions in watched queries

* added viewName override for tables

* update powersync-sqlite-core v0.1.6

* add Global sync locks in React Native

Author: stevensJourney

.changeset/chilled-poets-think.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-react-native': patch
+---
+
+Added global locks for syncing connections. Added warning when creating multiple Powersync instances.

.changeset/curly-peas-argue.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-common': minor
+---
+
+Added `viewName` option to Schema Table definitions. This allows for overriding a table's view name.

.changeset/eight-squids-peel.md

@@ -0,0 +1,7 @@
+---
+'@journeyapps/powersync-sdk-react-native': minor
+---
+
+Bumped powersync-sqlite-core to v0.1.6. dependant projects should:
+- Upgrade to `@journeyapps/react-native-quick-sqlite@1.1.1`
+- run `pod repo update && pod update` in the `ios` folder for updates to reflect.

.changeset/warm-foxes-act.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-common': patch
+---
+
+Improved table change updates to be throttled on the trailing edge. This prevents unnecessary query on both the leading and rising edge. 

apps/supabase-todolist

@@ -18,6 +18,11 @@ import { CrudEntry } from './sync/bucket/CrudEntry';
 import { mutexRunExclusive } from '../utils/mutex';
 import { BaseObserver } from '../utils/BaseObserver';
 import { EventIterator } from 'event-iterator';
+import { quoteIdentifier } from '../utils/strings';
+
+export interface DisconnectAndClearOptions {
+  clearLocal?: boolean;
+}
 
 export interface PowerSyncDatabaseOptions {
   schema: Schema;
@@ -57,6 +62,10 @@ export interface PowerSyncDBListener extends StreamingSyncImplementationListener
 
 const POWERSYNC_TABLE_MATCH = /(^ps_data__|^ps_data_local__)/;
 
+const DEFAULT_DISCONNECT_CLEAR_OPTIONS: DisconnectAndClearOptions = {
+  clearLocal: true
+};
+
 export const DEFAULT_WATCH_THROTTLE_MS = 30;
 
 export const DEFAULT_POWERSYNC_DB_OPTIONS = {
@@ -90,21 +99,23 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   protected bucketStorageAdapter: BucketStorageAdapter;
   private syncStatusListenerDisposer?: () => void;
   protected _isReadyPromise: Promise<void>;
+  protected _schema: Schema;
 
   constructor(protected options: PowerSyncDatabaseOptions) {
     super();
     this.bucketStorageAdapter = this.generateBucketStorageAdapter();
     this.closed = true;
     this.currentStatus = null;
     this.options = { ...DEFAULT_POWERSYNC_DB_OPTIONS, ...options };
+    this._schema = options.schema;
     this.ready = false;
     this.sdkVersion = '';
     // Start async init
     this._isReadyPromise = this.initialize();
   }
 
   get schema() {
-    return this.options.schema;
+    return this._schema;
   }
 
   protected get database() {
@@ -145,13 +156,32 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   protected async initialize() {
     await this._initialize();
     await this.bucketStorageAdapter.init();
-    await this.database.execute('SELECT powersync_replace_schema(?)', [JSON.stringify(this.schema.toJSON())]);
     const version = await this.options.database.execute('SELECT powersync_rs_version()');
     this.sdkVersion = version.rows?.item(0)['powersync_rs_version()'] ?? '';
+    await this.updateSchema(this.options.schema);
     this.ready = true;
     this.iterateListeners((cb) => cb.initialized?.());
   }
 
+  async updateSchema(schema: Schema) {
+    if (this.abortController) {
+      throw new Error('Cannot update schema while connected');
+    }
+
+    /**
+     * TODO
+     * Validations only show a warning for now.
+     * The next major release should throw an exception.
+     */
+    try {
+      schema.validate();
+    } catch (ex) {
+      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())]);
+  }
+
   /**
    * Queues a CRUD upload when internal CRUD tables have been updated
    */
@@ -208,24 +238,31 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    *  The database can still be queried after this is called, but the tables
    *  would be empty.
    */
-  async disconnectAndClear() {
+  async disconnectAndClear(options = DEFAULT_DISCONNECT_CLEAR_OPTIONS) {
     await this.disconnect();
 
+    const { clearLocal } = options;
+
     // TODO DB name, verify this is necessary with extension
     await this.database.writeTransaction(async (tx) => {
-      await tx.execute(`DELETE FROM ${PSInternalTable.OPLOG} WHERE 1`);
-      await tx.execute(`DELETE FROM ${PSInternalTable.CRUD} WHERE 1`);
-      await tx.execute(`DELETE FROM ${PSInternalTable.BUCKETS} WHERE 1`);
+      await tx.execute(`DELETE FROM ${PSInternalTable.OPLOG}`);
+      await tx.execute(`DELETE FROM ${PSInternalTable.CRUD}`);
+      await tx.execute(`DELETE FROM ${PSInternalTable.BUCKETS}`);
+
+      const tableGlob = clearLocal ? 'ps_data_*' : 'ps_data__*';
 
       const existingTableRows = await tx.execute(
-        "SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data_*'"
+        `
+      SELECT name FROM sqlite_master WHERE type='table' AND name GLOB ?
+      `,
+        [tableGlob]
       );
 
       if (!existingTableRows.rows.length) {
         return;
       }
       for (const row of existingTableRows.rows._array) {
-        await tx.execute(`DELETE FROM ${row.name} WHERE 1`);
+        await tx.execute(`DELETE FROM ${quoteIdentifier(row.name)} WHERE 1`);
       }
     });
   }
@@ -499,15 +536,19 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     const throttleMs = options.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS;
 
     return new EventIterator<WatchOnChangeEvent>((eventOptions) => {
-      const flushTableUpdates = _.throttle(async () => {
-        const intersection = _.intersection(watchedTables, throttledTableUpdates);
-        if (intersection.length) {
-          eventOptions.push({
-            changedTables: intersection
-          });
-        }
-        throttledTableUpdates = [];
-      }, throttleMs);
+      const flushTableUpdates = _.throttle(
+        async () => {
+          const intersection = _.intersection(watchedTables, throttledTableUpdates);
+          if (intersection.length) {
+            eventOptions.push({
+              changedTables: intersection
+            });
+          }
+          throttledTableUpdates = [];
+        },
+        throttleMs,
+        { leading: false, trailing: true }
+      );
 
       const dispose = this.database.registerListener({
         tablesUpdated: async (update) => {

packages/powersync-sdk-common/src/client/AbstractPowerSyncDatabase.ts

@@ -1,3 +1,4 @@
+import Logger from 'js-logger';
 import { DBAdapter } from '../db/DBAdapter';
 import { Schema } from '../db/schema/Schema';
 import { AbstractPowerSyncDatabase, PowerSyncDatabaseOptions } from './AbstractPowerSyncDatabase';
@@ -15,7 +16,9 @@ export interface PowerSyncOpenFactoryOptions extends Partial<PowerSyncDatabaseOp
 }
 
 export abstract class AbstractPowerSyncDatabaseOpenFactory {
-  constructor(protected options: PowerSyncOpenFactoryOptions) {}
+  constructor(protected options: PowerSyncOpenFactoryOptions) {
+    options.logger = options.logger ?? Logger.get(`PowerSync ${this.options.dbFilename}`);
+  }
 
   get schema() {
     return this.options.schema;

packages/powersync-sdk-common/src/client/AbstractPowerSyncOpenFactory.ts

@@ -33,11 +33,16 @@ export interface LockOptions<T> {
 
 export interface AbstractStreamingSyncImplementationOptions {
   adapter: BucketStorageAdapter;
-  remote: AbstractRemote;
   uploadCrud: () => Promise<void>;
+  crudUploadThrottleMs?: number;
+  /**
+   * An identifier for which PowerSync DB this sync implementation is
+   * linked to. Most commonly DB name, but not restricted to DB name.
+   */
+  identifier?: string;
   logger?: ILogger;
+  remote: AbstractRemote;
   retryDelayMs?: number;
-  crudUploadThrottleMs?: number;
 }
 
 export interface StreamingSyncImplementationListener extends BaseListener {

packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -3,6 +3,12 @@ import type { Table } from './Table';
 export class Schema {
   constructor(public tables: Table[]) {}
 
+  validate() {
+    for (const table of this.tables) {
+      table.validate();
+    }
+  }
+
   toJSON() {
     return {
       tables: this.tables.map((t) => t.toJSON())

packages/powersync-sdk-common/src/db/schema/Schema.ts

@@ -1,12 +1,17 @@
+import _ from 'lodash';
 import { Column } from '../Column';
 import type { Index } from './Index';
 
 export interface TableOptions {
+  /**
+   * The synced table name, matching sync rules
+   */
   name: string;
   columns: Column[];
   indexes?: Index[];
   localOnly?: boolean;
   insertOnly?: boolean;
+  viewName?: string;
 }
 
 export const DEFAULT_TABLE_OPTIONS: Partial<TableOptions> = {
@@ -15,6 +20,8 @@ export const DEFAULT_TABLE_OPTIONS: Partial<TableOptions> = {
   localOnly: false
 };
 
+export const InvalidSQLCharacters = /[\"\'%,\.#\s\[\]]/;
+
 export class Table {
   protected options: TableOptions;
 
@@ -34,6 +41,14 @@ export class Table {
     return this.options.name;
   }
 
+  get viewNameOverride() {
+    return this.options.viewName;
+  }
+
+  get viewName() {
+    return this.viewNameOverride || this.name;
+  }
+
   get columns() {
     return this.options.columns;
   }
@@ -59,13 +74,57 @@ export class Table {
   }
 
   get validName() {
-    // TODO verify
-    return !/[\"\'%,\.#\s\[\]]/.test(this.name);
+    return _.chain([this.name, this.viewNameOverride])
+      .compact()
+      .every((name) => !InvalidSQLCharacters.test(name))
+      .value();
+  }
+
+  validate() {
+    if (InvalidSQLCharacters.test(this.name)) {
+      throw new Error(`Invalid characters in table name: ${this.name}`);
+    } else if (this.viewNameOverride && InvalidSQLCharacters.test(this.viewNameOverride!)) {
+      throw new Error(`
+          Invalid characters in view name: ${this.viewNameOverride}`);
+    }
+
+    const columnNames = new Set<string>();
+    columnNames.add('id');
+    for (const column of this.columns) {
+      const { name: columnName } = column;
+      if (column.name == 'id') {
+        throw new Error(`${this.name}: id column is automatically added, custom id columns are not supported`);
+      } else if (columnNames.has(columnName)) {
+        throw new Error(`Duplicate column ${columnName}`);
+      } else if (InvalidSQLCharacters.test(columnName)) {
+        throw new Error(`Invalid characters in column name: $name.${column}`);
+      }
+      columnNames.add(columnName);
+    }
+
+    const indexNames = new Set<string>();
+
+    for (const index of this.indexes) {
+      if (indexNames.has(index.name)) {
+        throw new Error(`Duplicate index $name.${index}`);
+      } else if (InvalidSQLCharacters.test(index.name)) {
+        throw new Error(`Invalid characters in index name: $name.${index}`);
+      }
+
+      for (const column of index.columns) {
+        if (!columnNames.has(column.name)) {
+          throw new Error(`Column ${column.name} not found for index ${index.name}`);
+        }
+      }
+
+      indexNames.add(index.name);
+    }
   }
 
   toJSON() {
     return {
       name: this.name,
+      view_name: this.viewName,
       local_only: this.localOnly,
       insert_only: this.insertOnly,
       columns: this.columns.map((c) => c.toJSON()),