feat: 🎸 Add a shared service between web workers (#2932)

* feat: 🎸 Add a shared service between web workers

✅ Closes: https://hashicorp.atlassian.net/browse/ICU-17409

addons/api/addon/services/sqlite-db.js

@@ -21,7 +21,7 @@ export default class SqliteDbService extends Service {
       return;
     }
 
-    // In production, our JS files will be fingerprinted with am MD5 hash
+    // In production, our JS files will be fingerprinted with an MD5 hash
     // so we need a way to map the worker file name to be able to find it.
     //
     // Ember is automagically replacing any URL path with the fingerprinted
@@ -31,7 +31,7 @@ export default class SqliteDbService extends Service {
     // https://github.com/ember-cli/broccoli-asset-rewrite/issues/66
     // and will not work properly if it's part of a template string or interpolated.
     // Therefore, we need the URL path in a string somewhere to be replaced properly.
-    this.webWorker = new Worker('/workers/web-worker.js', {
+    this.webWorker = new Worker('/workers/sqlite-worker.js', {
       type: 'module',
     });
 

addons/api/addon/workers/sqlite-worker.js

@@ -0,0 +1,45 @@
+/**
+ * Copyright (c) HashiCorp, Inc.
+ * SPDX-License-Identifier: BUSL-1.1
+ */
+
+import { PWBWorker } from 'promise-worker-bi';
+import sqlite3InitModule from '@sqlite.org/sqlite-wasm';
+import sharedServiceInit from './utils/shared-service';
+
+const methods = {
+  initializeSQLite: async () => {
+    try {
+      const sqlite3 = await sqlite3InitModule({
+        print: console.log,
+        printErr: console.error,
+      });
+      console.log(`SQLite Version: ${sqlite3?.version.libVersion}`);
+      const poolUtil = await sqlite3.installOpfsSAHPoolVfs();
+      new poolUtil.OpfsSAHPoolDb('/boundary');
+    } catch (err) {
+      console.error('Initialization error:', err.name, err.message);
+    }
+  },
+};
+
+const promiseWorker = new PWBWorker();
+promiseWorker.register(async ({ method, payload }) => {
+  await sharedService.status.ready;
+
+  return sharedService.proxy[method](payload);
+});
+
+const initializeOnProviderChange = async (isProvider) => {
+  if (!isProvider) {
+    return;
+  }
+  // Initialize the SQLite database when we become a provider
+  await sharedService.proxy.initializeSQLite();
+};
+
+const sharedService = sharedServiceInit(
+  'sqlite',
+  methods,
+  initializeOnProviderChange,
+);

addons/api/addon/workers/utils/shared-service.js

@@ -0,0 +1,269 @@
+/**
+ * Copyright (c) HashiCorp, Inc.
+ * SPDX-License-Identifier: BUSL-1.1
+ */
+
+import { v4 as uuidv4 } from 'uuid';
+
+let clientChannel = null;
+let commonChannel = null;
+const requestsInFlight = new Map();
+
+export default (serviceName, target, onProviderChange) => {
+  commonChannel = new BroadcastChannel(
+    `shared-service-common-channel-${serviceName}`,
+  );
+
+  let { promise: ready, resolve: readyResolve } = Promise.withResolvers();
+  const status = {
+    ready,
+    isServiceProvider: false,
+  };
+
+  const proxy = getProxy(target, status);
+
+  // This lock is used to ensure that only one tab can become the provider with an indefinite lock.
+  navigator.locks.request(serviceName, { mode: 'exclusive' }, async () => {
+    // If a client was promoted to provider, reset the ready status and resolver
+    if (readyResolve === null) {
+      const { promise: newReady, resolve: newResolve } =
+        Promise.withResolvers();
+      status.ready = newReady;
+      readyResolve = newResolve;
+    }
+
+    status.isServiceProvider = true;
+    await onBecomeProvider({
+      serviceName,
+      onProviderChange,
+      target,
+      status,
+    });
+    readyResolve();
+    readyResolve = null;
+    await new Promise(() => {});
+  });
+
+  (async () => {
+    let providerExists = false;
+
+    // Make sure a lock was acquired which means we have a provider
+    while (!providerExists) {
+      const locks = await navigator.locks.query();
+      providerExists = locks.held?.some((lock) => lock.name === serviceName);
+    }
+
+    if (!status.isServiceProvider) {
+      await onNotProvider({
+        serviceName,
+        onProviderChange,
+        status,
+      });
+      readyResolve();
+      readyResolve = null;
+    }
+  })();
+
+  return { proxy, status };
+};
+
+const getClientBroadcastChannelName = (clientId, serviceName) =>
+  `${clientId}-${serviceName}`;
+
+// Generate a random ID for the tab to use as a lock name to hold indefinitely until the tab is closed
+const getClientId = () => {
+  const clientId = uuidv4();
+
+  // Keep the lock until this context is destroyed (which means the tab was closed).
+  navigator.locks.request(clientId, { mode: 'exclusive' }, async () => {
+    await new Promise(() => {});
+  });
+
+  return clientId;
+};
+
+// Sends a message to the provider to call the method on behalf of a client.
+const requestFromProvider = (uuid, resolve, reject, method, args) => {
+  const onRequestListener = getOnRequestListener(uuid, resolve, reject);
+
+  // Listen for a response to know we finished and remove from the list of requests in flight
+  clientChannel?.addEventListener('message', onRequestListener);
+  clientChannel?.postMessage({
+    id: uuid,
+    type: 'request',
+    method,
+    args,
+  });
+};
+
+// Return the listener used for responding to requests. Only listens for response messages.
+const getOnRequestListener = (uuid, resolve, reject) => {
+  const listener = ({ data }) => {
+    const { id, type } = data;
+    if (id !== uuid || type === 'request') {
+      return;
+    }
+
+    requestsInFlight.delete(uuid);
+    clientChannel?.removeEventListener('message', listener);
+
+    if (data.error) {
+      reject(data.error);
+    }
+
+    resolve(data.result);
+  };
+
+  return listener;
+};
+
+// When we become a client instead of a provider we do two things:
+// 1. Register ourselves with the provider and wait for a response that says we registered.
+// 2. Setup an event listener to know when a provider has changed.
+const onNotProvider = async ({ serviceName, onProviderChange, status }) => {
+  const clientId = getClientId();
+  clientChannel = new BroadcastChannel(
+    getClientBroadcastChannelName(clientId, serviceName),
+  );
+
+  const register = async () =>
+    new Promise((resolve) => {
+      const onRegisteredListener = ({ data }) => {
+        if (data.clientId === clientId && data.type === 'registered') {
+          commonChannel.removeEventListener('message', onRegisteredListener);
+          resolve();
+        }
+      };
+
+      // Add an event listener to check for the registration response
+      commonChannel.addEventListener('message', onRegisteredListener);
+      commonChannel.postMessage({ type: 'register', clientId });
+    });
+
+  commonChannel.addEventListener('message', async ({ data }) => {
+    // Check if this was a provider change. If so and we're not the new provider, re-register.
+    if (data.type === 'providerChange' && !status.isServiceProvider) {
+      await onProviderChange?.(status.isServiceProvider);
+      await register();
+
+      // If there were requests in flight when the provider was changed,
+      // we need to resend them to the new provider.
+      if (requestsInFlight.size > 0) {
+        requestsInFlight.forEach(({ method, args, resolve, reject }, uuid) => {
+          requestFromProvider(uuid, resolve, reject, method, args);
+        });
+      }
+    }
+  });
+
+  await register();
+};
+
+// When we become the provider, we do two things:
+// 1. Set an event listener to check for clients trying to register with us. As part of registration,
+//    we also setup a listener with the individual client to listen for requests.
+// 2. Send a message to all tabs that we've now become the provider
+const onBecomeProvider = async ({
+  serviceName,
+  onProviderChange,
+  target,
+  status,
+}) => {
+  // Set an event listener for other tabs to register with the provider tab
+  commonChannel.addEventListener('message', async (event) => {
+    const { clientId, type } = event.data;
+    if (type !== 'register') {
+      return;
+    }
+
+    const clientChannel = new BroadcastChannel(
+      getClientBroadcastChannelName(clientId, serviceName),
+    );
+
+    // Acquire another lock that immediately is waiting for the
+    // initial lock on the client tab to be resolved to handle cleanup.
+    navigator.locks.request(clientId, { mode: 'exclusive' }, async () => {
+      // The client tab was closed, clean up.
+      clientChannel.close();
+    });
+
+    // Setup an event listener for communication between the provider and subscriber to respond to requests
+    clientChannel.addEventListener('message', async ({ data }) => {
+      if (data.type === 'response') {
+        return;
+      }
+      const { method, args, id } = data;
+
+      let result, error;
+      try {
+        result = await target[method](...args);
+      } catch (e) {
+        // Error objects won't be properly serialized, we need to reconstruct an error like object
+        error =
+          e instanceof Error
+            ? Object.fromEntries(
+                Object.getOwnPropertyNames(e).map((k) => [k, e[k]]),
+              )
+            : e;
+      }
+
+      clientChannel.postMessage({
+        id,
+        type: 'response',
+        result,
+        error,
+        method,
+      });
+    });
+
+    // Send back a message to the client that we finished registering
+    commonChannel.postMessage({ type: 'registered', clientId, serviceName });
+  });
+
+  await onProviderChange?.(status.isServiceProvider);
+  commonChannel.postMessage({ type: 'providerChange', serviceName });
+
+  // If we became the new provider and there were requests in flight, request it directly again.
+  if (requestsInFlight.size > 0) {
+    await Promise.all(
+      Array.from(
+        requestsInFlight,
+        async ([uuid, { method, args, resolve, reject }]) => {
+          try {
+            const result = await target[method](...args);
+            resolve(result);
+          } catch (error) {
+            reject(error);
+          } finally {
+            requestsInFlight.delete(uuid);
+          }
+        },
+      ),
+    );
+  }
+};
+
+// Returns a proxy that we will have all caller methods go through.
+// If we're the provider, we just call the method directly. If we're not the provider, we'll
+// send the request through our established client channel to have the provider respond with the data.
+const getProxy = (target, status) =>
+  new Proxy(target, {
+    get: (target, method) => {
+      if (method === 'then' || method === 'catch' || method === 'finally') {
+        // Return undefined for these methods to allow promise chaining to work correctly
+        return;
+      }
+
+      return async (...args) => {
+        if (status.isServiceProvider) {
+          return await target[method](...args);
+        }
+
+        return new Promise((resolve, reject) => {
+          const uuid = uuidv4();
+          requestFromProvider(uuid, resolve, reject, method, args);
+          requestsInFlight.set(uuid, { method, args, resolve, reject });
+        });
+      };
+    },
+  });

addons/api/addon/workers/web-worker.js

@@ -1,31 +0,0 @@
-/**
- * Copyright (c) HashiCorp, Inc.
- * SPDX-License-Identifier: BUSL-1.1
- */
-
-import { PWBWorker } from 'promise-worker-bi';
-import sqlite3InitModule from '@sqlite.org/sqlite-wasm';
-
-const promiseWorker = new PWBWorker();
-promiseWorker.register((message) => {
-  console.log(`Received from main: ${message}`);
-});
-
-const initializeSQLite = async () => {
-  try {
-    const sqlite3 = await sqlite3InitModule({
-      print: console.log,
-      printErr: console.error,
-    });
-    console.log(`SQLite Version: ${sqlite3?.version.libVersion}`);
-    const poolUtil = await sqlite3.installOpfsSAHPoolVfs();
-    const db = new poolUtil.OpfsSAHPoolDb('/boundary');
-    console.log(db);
-  } catch (err) {
-    console.error('Initialization error:', err.name, err.message);
-  }
-};
-
-(async () => {
-  await initializeSQLite();
-})();

addons/api/broccoli-plugins/build-workers.js

@@ -31,9 +31,11 @@ module.exports = class BuildWorkers extends Plugin {
     let workers = {};
     let dir = fs.readdirSync(inputPath);
 
-    dir.forEach((name) => {
-      workers[name] = path.join(inputPath, name);
-    });
+    dir
+      .filter((name) => name.endsWith('-worker.js'))
+      .forEach((name) => {
+        workers[name] = path.join(inputPath, name);
+      });
 
     return workers;
   }

addons/api/package.json

@@ -42,7 +42,8 @@
     "esbuild": "^0.25.4",
     "miragejs": "^0.1.48",
     "promise-worker-bi": "^5.0.1",
-    "tracked-built-ins": "^4.0.0"
+    "tracked-built-ins": "^4.0.0",
+    "uuid": "^11.0.3"
   },
   "devDependencies": {
     "@babel/eslint-parser": "^7.25.1",