cleanup config sharing

Author: stevensJourney

packages/web/src/db/adapters/AsyncDatabaseConnection.ts

@@ -17,14 +17,15 @@ export type OnTableChangeCallback = (event: BatchedUpdateNotification) => void;
  * An async Database connection which provides basic async SQL methods.
  * This is usually a proxied through a web worker.
  */
-export interface AsyncDatabaseConnection {
+export interface AsyncDatabaseConnection<Config extends ResolvedWebSQLOpenOptions = ResolvedWebSQLOpenOptions> {
   init(): Promise<void>;
   close(): Promise<void>;
   execute(sql: string, params?: any[]): Promise<ProxiedQueryResult>;
   executeBatch(sql: string, params?: any[]): Promise<ProxiedQueryResult>;
   registerOnTableChange(callback: OnTableChangeCallback): Promise<() => void>;
+  getConfig(): Promise<Config>;
 }
 
-export type OpenAsyncDatabaseConnection<Options extends ResolvedWebSQLOpenOptions = ResolvedWebSQLOpenOptions> = (
-  options: Options
+export type OpenAsyncDatabaseConnection<Config extends ResolvedWebSQLOpenOptions = ResolvedWebSQLOpenOptions> = (
+  config: Config
 ) => AsyncDatabaseConnection;

packages/web/src/db/adapters/LockedAsyncDatabaseAdapter.ts

@@ -13,6 +13,7 @@ import { getNavigatorLocks } from '../..//shared/navigator';
 import { AsyncDatabaseConnection } from './AsyncDatabaseConnection';
 import { SharedConnectionWorker, WebDBAdapter } from './WebDBAdapter';
 import { WorkerWrappedAsyncDatabaseConnection } from './WorkerWrappedAsyncDatabaseConnection';
+import { ResolvedWebSQLOpenOptions } from './web-sql-flags';
 
 /**
  * @internal
@@ -44,6 +45,7 @@ export class LockedAsyncDatabaseAdapter
   protected initPromise: Promise<void>;
   private _db: AsyncDatabaseConnection | null = null;
   protected _disposeTableChangeListener: (() => void) | null = null;
+  private _config: ResolvedWebSQLOpenOptions | null = null;
 
   constructor(protected options: LockedAsyncDatabaseAdapterOptions) {
     super();
@@ -83,23 +85,6 @@ export class LockedAsyncDatabaseAdapter
     return this._dbIdentifier;
   }
 
-  async shareConnection(): Promise<SharedConnectionWorker> {
-    if (false == this._db instanceof WorkerWrappedAsyncDatabaseConnection) {
-      throw new Error(`Only worker connections can be shared`);
-    }
-    return this._db.shareConnection();
-  }
-
-  /**
-   * Registers a table change notification callback with the base database.
-   * This can be extended by custom implementations in order to handle proxy events.
-   */
-  protected async registerOnChangeListener(db: AsyncDatabaseConnection) {
-    this._disposeTableChangeListener = await db.registerOnTableChange((event) => {
-      this.iterateListeners((cb) => cb.tablesUpdated?.(event));
-    });
-  }
-
   /**
    * Init is automatic, this helps catch errors or explicitly await initialization
    */
@@ -110,15 +95,40 @@ export class LockedAsyncDatabaseAdapter
   protected async _init() {
     this._db = await this.options.openConnection();
     await this._db.init();
+    this._config = await this._db.getConfig();
     await this.registerOnChangeListener(this._db);
     this.iterateListeners((cb) => cb.initialized?.());
   }
 
+  getConfiguration(): ResolvedWebSQLOpenOptions {
+    if (!this._config) {
+      throw new Error(`Cannot get config before initialization is completed`);
+    }
+    return this._config;
+  }
+
   protected async waitForInitialized() {
     // Awaiting this will expose errors on function calls like .execute etc
     await this.initPromise;
   }
 
+  async shareConnection(): Promise<SharedConnectionWorker> {
+    if (false == this._db instanceof WorkerWrappedAsyncDatabaseConnection) {
+      throw new Error(`Only worker connections can be shared`);
+    }
+    return this._db.shareConnection();
+  }
+
+  /**
+   * Registers a table change notification callback with the base database.
+   * This can be extended by custom implementations in order to handle proxy events.
+   */
+  protected async registerOnChangeListener(db: AsyncDatabaseConnection) {
+    this._disposeTableChangeListener = await db.registerOnTableChange((event) => {
+      this.iterateListeners((cb) => cb.tablesUpdated?.(event));
+    });
+  }
+
   /**
    * This is currently a no-op on web
    */

packages/web/src/db/adapters/WebDBAdapter.ts

@@ -1,4 +1,5 @@
 import { DBAdapter } from '@powersync/common';
+import { ResolvedWebSQLOpenOptions } from './web-sql-flags';
 
 export type SharedConnectionWorker = {
   identifier: string;
@@ -10,4 +11,10 @@ export interface WebDBAdapter extends DBAdapter {
    * Get a MessagePort which can be used to share the internals of this connection.
    */
   shareConnection(): Promise<SharedConnectionWorker>;
+
+  /**
+   * Get the config options used to open this connection.
+   * This is useful for sharing connections.
+   */
+  getConfiguration(): ResolvedWebSQLOpenOptions;
 }

packages/web/src/db/adapters/WorkerWrappedAsyncDatabaseConnection.ts

@@ -68,4 +68,8 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
   executeBatch(sql: string, params?: any[]): Promise<ProxiedQueryResult> {
     return this.baseConnection.executeBatch(sql, params);
   }
+
+  getConfig(): Promise<ResolvedWebSQLOpenOptions> {
+    return this.baseConnection.getConfig();
+  }
 }

packages/web/src/db/adapters/wa-sqlite/WASQLiteConnection.ts

@@ -2,7 +2,7 @@ import * as SQLite from '@journeyapps/wa-sqlite';
 import { BaseObserver, BatchedUpdateNotification } from '@powersync/common';
 import { Mutex } from 'async-mutex';
 import { AsyncDatabaseConnection, OnTableChangeCallback, ProxiedQueryResult } from '../AsyncDatabaseConnection';
-import { ResolvedWebSQLOpenOptions } from '../web-sql-flags';
+import { ResolvedWASQLiteOpenFactoryOptions } from './WASQLiteOpenFactory';
 
 /**
  * List of currently tested virtual filesystems
@@ -45,13 +45,6 @@ export type WASQLiteModuleFactory = (
   options: WASQLiteModuleFactoryOptions
 ) => Promise<{ module: SQLiteModule; vfs: SQLiteVFS }>;
 
-/**
- * @internal
- */
-export interface WASQLiteOpenOptions extends ResolvedWebSQLOpenOptions {
-  vfs?: WASQLiteVFS;
-}
-
 /**
  * @internal
  */
@@ -106,7 +99,10 @@ export const DEFAULT_MODULE_FACTORIES = {
  * WA-SQLite connection which directly interfaces with WA-SQLite.
  * This is usually instantiated inside a worker.
  */
-export class WASqliteConnection extends BaseObserver<WASQLiteConnectionListener> implements AsyncDatabaseConnection {
+export class WASqliteConnection
+  extends BaseObserver<WASQLiteConnectionListener>
+  implements AsyncDatabaseConnection<ResolvedWASQLiteOpenFactoryOptions>
+{
   private _sqliteAPI: SQLiteAPI | null = null;
   private _dbP: number | null = null;
   private _moduleFactory: WASQLiteModuleFactory;
@@ -121,14 +117,14 @@ export class WASqliteConnection extends BaseObserver<WASQLiteConnectionListener>
    */
   protected connectionId: number;
 
-  constructor(protected options: WASQLiteOpenOptions) {
+  constructor(protected options: ResolvedWASQLiteOpenFactoryOptions) {
     super();
     this.updatedTables = new Set();
     this.updateTimer = null;
     this.broadcastChannel = null;
     this.connectionId = new Date().valueOf() + Math.random();
     this.statementMutex = new Mutex();
-    this._moduleFactory = DEFAULT_MODULE_FACTORIES[this.options.vfs ?? WASQLiteVFS.IDBBatchAtomicVFS];
+    this._moduleFactory = DEFAULT_MODULE_FACTORIES[this.options.vfs];
   }
 
   protected get sqliteAPI() {
@@ -185,6 +181,7 @@ export class WASqliteConnection extends BaseObserver<WASQLiteConnectionListener>
     this._sqliteAPI = await this.openSQLiteAPI();
     await this.openDB();
     this.registerBroadcastListeners();
+    await this.executeSingleStatement(`PRAGMA temp_store = ${this.options.temporaryStorage};`);
 
     this.sqliteAPI.update_hook(this.dbP, (updateType: number, dbName: string | null, tableName: string | null) => {
       if (!tableName) {
@@ -195,6 +192,10 @@ export class WASqliteConnection extends BaseObserver<WASQLiteConnectionListener>
     });
   }
 
+  async getConfig(): Promise<ResolvedWASQLiteOpenFactoryOptions> {
+    return this.options;
+  }
+
   fireUpdates() {
     this.updateTimer = null;
     const event: BatchedUpdateNotification = { tables: [...this.updatedTables], groupedUpdates: {}, rawUpdates: [] };

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -97,10 +97,11 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
   protected messagePort: MessagePort;
 
   protected isInitialized: Promise<void>;
+  protected dbAdapter: WebDBAdapter;
 
   constructor(options: SharedWebStreamingSyncImplementationOptions) {
     super(options);
-
+    this.dbAdapter = options.db;
     /**
      * Configure or connect to the shared sync worker.
      * This worker will manage all syncing operations remotely.
@@ -147,7 +148,7 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     const flags = { ...this.webOptions.flags, workers: undefined };
 
     this.isInitialized = this.syncManager.setParams({
-      dbName: this.options.identifier!,
+      dbParams: this.dbAdapter.getConfiguration(),
       streamOptions: {
         crudUploadThrottleMs,
         identifier,

packages/web/src/worker/db/WASQLiteDB.worker.ts

@@ -5,7 +5,8 @@
 import '@journeyapps/wa-sqlite';
 import * as Comlink from 'comlink';
 import { AsyncDatabaseConnection } from '../../db/adapters/AsyncDatabaseConnection';
-import { WASQLiteOpenOptions, WASqliteConnection } from '../../db/adapters/wa-sqlite/WASQLiteConnection';
+import { WASqliteConnection } from '../../db/adapters/wa-sqlite/WASQLiteConnection';
+import { ResolvedWASQLiteOpenFactoryOptions } from '../../db/adapters/wa-sqlite/WASQLiteOpenFactory';
 import { getNavigatorLocks } from '../../shared/navigator';
 
 /**
@@ -22,10 +23,11 @@ const OPEN_DB_LOCK = 'open-wasqlite-db';
 
 let nextClientId = 1;
 
-const openWorkerConnection = async (options: WASQLiteOpenOptions): Promise<AsyncDatabaseConnection> => {
+const openWorkerConnection = async (options: ResolvedWASQLiteOpenFactoryOptions): Promise<AsyncDatabaseConnection> => {
   const connection = new WASqliteConnection(options);
   return {
     init: Comlink.proxy(() => connection.init()),
+    getConfig: Comlink.proxy(() => connection.getConfig()),
     close: Comlink.proxy(() => connection.close()),
     execute: Comlink.proxy(async (sql: string, params?: any[]) => connection.execute(sql, params)),
     executeBatch: Comlink.proxy(async (sql: string, params?: any[]) => connection.executeBatch(sql, params)),
@@ -36,7 +38,7 @@ const openWorkerConnection = async (options: WASQLiteOpenOptions): Promise<Async
   };
 };
 
-const openDBShared = async (options: WASQLiteOpenOptions): Promise<AsyncDatabaseConnection> => {
+const openDBShared = async (options: ResolvedWASQLiteOpenFactoryOptions): Promise<AsyncDatabaseConnection> => {
   // Prevent multiple simultaneous opens from causing race conditions
   return getNavigatorLocks().request(OPEN_DB_LOCK, async () => {
     const clientId = nextClientId++;