share worker for OPFS

Author: stevensJourney

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

@@ -4,5 +4,5 @@ export interface WorkerDBAdapter extends DBAdapter {
   /**
    * Get a MessagePort which can be used to share the internals of this connection.
    */
-  getMessagePort(): MessagePort | Worker;
+  getMessagePort(): Promise<MessagePort>;
 }

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

@@ -1,3 +1,4 @@
+import * as Comlink from 'comlink';
 import { LockedAsyncDatabaseAdapter, LockedAsyncDatabaseAdapterOptions } from './LockedAsyncDatabaseAdapter';
 import { ProxiedAsyncDatabaseConnection } from './ProxiedAsyncDatabaseConnection';
 import { WorkerDBAdapter } from './WorkerDBAdapter';
@@ -25,7 +26,16 @@ export class WorkerLockedAsyncDatabaseAdapter extends LockedAsyncDatabaseAdapter
     this._messagePort = options.messagePort;
   }
 
-  getMessagePort(): MessagePort | Worker {
+  async getMessagePort(): Promise<MessagePort> {
+    if (this._messagePort instanceof Worker) {
+      // We can't transfer a Worker instance, need a MessagePort
+      // Comlink provides a nice utility for exposing a MessagePort
+      // from a Worker
+      const temp = Comlink.wrap(this._messagePort);
+      const newPort = await temp[Comlink.createEndpoint]();
+      return newPort;
+    }
+
     return this._messagePort;
   }
 }

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

@@ -21,14 +21,14 @@ class SharedSyncClientProvider extends AbstractSharedSyncClientProvider {
   constructor(
     protected options: WebStreamingSyncImplementationOptions,
     public statusChanged: (status: SyncStatusOptions) => void,
-    protected dbWorkerPort: MessagePort | Worker
+    protected dbWorkerPort: Promise<MessagePort>
   ) {
     super();
   }
 
-  async getDBWorkerPort(): Promise<MessagePort | Worker> {
+  async getDBWorkerPort(): Promise<MessagePort> {
     // FIXME type error
-    const port = this.dbWorkerPort as MessagePort;
+    const port = (await this.dbWorkerPort) as MessagePort;
 
     // TODO this can only be done once. Throw an error if multiple attempts are made
     return Comlink.transfer(port, [port]);

packages/web/src/worker/sync/AbstractSharedSyncClientProvider.ts

@@ -7,7 +7,7 @@ export abstract class AbstractSharedSyncClientProvider {
   abstract fetchCredentials(): Promise<PowerSyncCredentials | null>;
   abstract uploadCrud(): Promise<void>;
   abstract statusChanged(status: SyncStatusOptions): void;
-  abstract getDBWorkerPort(): Promise<MessagePort | Worker>;
+  abstract getDBWorkerPort(): Promise<MessagePort>;
 
   abstract trace(...x: any[]): void;
   abstract debug(...x: any[]): void;

packages/web/src/worker/sync/SharedSyncImplementation.ts

@@ -279,7 +279,6 @@ export class SharedSyncImplementation
       // Ask for a new DB worker port handler
       const lastClient = this.ports[this.ports.length - 1];
       const workerPort = await lastClient.clientProvider.getDBWorkerPort();
-
       const locked = new LockedAsyncDatabaseAdapter({
         name: this.syncParams?.dbName!,
         openConnection: async () => {

packages/web/vitest.config.ts

@@ -1,9 +1,15 @@
-import wasm from 'vite-plugin-wasm';
-import topLevelAwait from 'vite-plugin-top-level-await';
 import path from 'path';
+import topLevelAwait from 'vite-plugin-top-level-await';
+import wasm from 'vite-plugin-wasm';
 import { defineConfig, UserConfigExport } from 'vitest/config';
 
 const config: UserConfigExport = {
+  server: {
+    headers: {
+      'Cross-Origin-Opener-Policy': 'same-origin',
+      'Cross-Origin-Embedder-Policy': 'require-corp'
+    }
+  },
   // This is only needed for local tests to resolve the package name correctly
   resolve: {
     alias: {