From b57b9458e450612fb20a20444cd0316ccd5fa40c Mon Sep 17 00:00:00 2001
From: Steven Vandevelde <steven@tokono.ma>
Date: Sat, 12 Jul 2025 09:50:35 +0200
Subject: [PATCH] feat: worker tunnel

---
 src/pages/configurator/input/_applet.astro  | 50 ++++++++++-----
 src/pages/core/types.d.ts                   |  2 +-
 src/pages/input/native-fs/_applet.astro     |  2 +-
 src/scripts/applet/common.ts                | 13 +++-
 src/scripts/common.ts                       | 62 ++++++++++++++++---
 src/scripts/configurator/input/common.ts    |  6 --
 src/scripts/configurator/input/constants.ts |  5 --
 src/scripts/configurator/input/types.d.ts   |  3 -
 src/scripts/configurator/input/worker.ts    | 67 ++++++---------------
 src/scripts/input/native-fs/worker.ts       |  4 +-
 src/scripts/input/opensubsonic/worker.ts    |  4 +-
 src/scripts/input/s3/worker.ts              |  4 +-
 src/scripts/processor/artwork/worker.ts     | 16 +++--
 13 files changed, 139 insertions(+), 99 deletions(-)
 delete mode 100644 src/scripts/configurator/input/common.ts
 delete mode 100644 src/scripts/configurator/input/constants.ts
 delete mode 100644 src/scripts/configurator/input/types.d.ts

diff --git a/src/pages/configurator/input/_applet.astro b/src/pages/configurator/input/_applet.astro
index d32a871..fc701a1 100644
--- a/src/pages/configurator/input/_applet.astro
+++ b/src/pages/configurator/input/_applet.astro
@@ -34,10 +34,9 @@
 <script>
   import type { Applet } from "@web-applets/sdk";
 
+  import type { Track } from "@applets/core/types";
   import type { Tasks } from "@scripts/configurator/input/worker";
-  import type { Scheme } from "@scripts/configurator/input/types";
-  import { CONNECTIONS } from "@scripts/configurator/input/constants";
-  import { applet, register } from "@scripts/applet/common";
+  import { applet, register, tunnel } from "@scripts/applet/common";
   import { endpoint, transfer } from "@scripts/common";
   import manifest from "./_manifest.json";
 
@@ -54,19 +53,42 @@
   // Register applet + worker
   const context = register({ worker });
 
+  // Applet connections
+  const input = {
+    "file+local": applet("/input/native-fs"),
+    opensubsonic: applet("/input/opensubsonic"),
+    s3: applet("/input/s3"),
+  };
+
+  // Provide tunnel to worker
+  tunnel(worker, input);
+
   ////////////////////////////////////////////
-  // CONNECTIONS
+  // ACTIONS
   ////////////////////////////////////////////
-  const connections: Record<Scheme, Promise<Applet>> = {
-    "file+local": applet(CONNECTIONS["file+local"], { context: self }),
-    opensubsonic: applet(CONNECTIONS.opensubsonic, { context: self }),
-    s3: applet(CONNECTIONS.s3, { context: self }),
+  const consult = async (fileUriOrScheme: string) => {
+    return await worker.consult(fileUriOrScheme);
+  };
+
+  const contextualize = async (cachedTracks: Track[]) => {
+    return await worker.contextualize(transfer(cachedTracks));
+  };
+
+  const groupConsult = async (tracks: Track[]) => {
+    return await worker.groupConsult(transfer(tracks));
+  };
+
+  const list = async (cachedTracks: Track[] = []) => {
+    return await worker.list(transfer(cachedTracks));
+  };
+
+  const resolve = async (args: { method: string; uri: string }) => {
+    return await worker.resolve(args);
   };
 
-  // Pass worker message ports to configurator worker
-  Object.entries(connections).forEach(([scheme, promise]) => {
-    promise.then((conn) => {
-      return worker.manageInputWorker(scheme as Scheme, transfer(conn.ports.worker));
-    });
-  });
+  context.setActionHandler("consult", consult);
+  context.setActionHandler("contextualize", contextualize);
+  context.setActionHandler("groupConsult", groupConsult);
+  context.setActionHandler("list", list);
+  context.setActionHandler("resolve", resolve);
 </script>
diff --git a/src/pages/core/types.d.ts b/src/pages/core/types.d.ts
index afbd528..306eb4f 100644
--- a/src/pages/core/types.d.ts
+++ b/src/pages/core/types.d.ts
@@ -20,7 +20,7 @@ export interface InputWorkerTasks {
   consult(fileUriOrScheme: string): Promise<Consult>;
   contextualize(tracks: Track[]): Promise<void>;
   groupConsult(tracks: Track[]): Promise<GroupConsult>;
-  list(cachedTracks: Track[] = []): Promise<Track[]>;
+  list(cachedTracks: Track[]): Promise<Track[]>;
   resolve({ method, uri }: { method: string; uri: string }): Promise<ResolvedUri>;
 }
 
diff --git a/src/pages/input/native-fs/_applet.astro b/src/pages/input/native-fs/_applet.astro
index a2e2f5c..a4317b6 100644
--- a/src/pages/input/native-fs/_applet.astro
+++ b/src/pages/input/native-fs/_applet.astro
@@ -55,7 +55,7 @@
     return await worker.list(transfer(cachedTracks));
   };
 
-  const resolve = async (args: { uri: string }) => {
+  const resolve = async (args: { method: string; uri: string }) => {
     return await worker.resolve(args);
   };
 
diff --git a/src/scripts/applet/common.ts b/src/scripts/applet/common.ts
index 366baa3..d3cd5e1 100644
--- a/src/scripts/applet/common.ts
+++ b/src/scripts/applet/common.ts
@@ -88,6 +88,17 @@ export async function applet<D>(
   return applet;
 }
 
+export function tunnel(
+  worker: Comlink.Remote<WorkerTasks>,
+  connections: Record<string, Applet | Promise<Applet>>,
+) {
+  Object.entries(connections).forEach(([scheme, promise]) => {
+    Promise.resolve(promise).then((conn) => {
+      return worker._manage(scheme, transfer(conn.ports.worker));
+    });
+  });
+}
+
 ////////////////////////////////////////////
 // 🪟 Applet registration
 ////////////////////////////////////////////
@@ -289,7 +300,7 @@ export function register<DataType = any>(
   if (options.worker !== undefined)
     context.scope.onworkerport = (event) => {
       if (!event.port) return;
-      options.worker?.listenForActions(transfer(event.port));
+      options.worker?._listen(transfer(event.port));
     };
 
   return context;
diff --git a/src/scripts/common.ts b/src/scripts/common.ts
index 55ee013..2ed4ed6 100644
--- a/src/scripts/common.ts
+++ b/src/scripts/common.ts
@@ -13,7 +13,8 @@ export const SharedWorker = globalThis.SharedWorker;
 ////////////////////////////////////////////
 
 export type WorkerTasks = {
-  listenForActions: ReturnType<typeof handleWorkerActions>;
+  _listen: ReturnType<typeof _listen>;
+  _manage: ReturnType<typeof _manage>;
 };
 
 ////////////////////////////////////////////
@@ -95,6 +96,16 @@ export function inIframe() {
   return window.self !== window.top;
 }
 
+export function initialConnections<C extends Record<string, any>>(ids: string[]) {
+  const connections: Record<string, PromiseWithResolvers<Comlink.Remote<C>>> = {};
+
+  ids.forEach((c) => {
+    connections[c] = Promise.withResolvers<Comlink.Remote<C>>();
+  });
+
+  return connections;
+}
+
 export function isPrimitive(test: unknown) {
   return test !== Object(test);
 }
@@ -107,14 +118,29 @@ export function jsonEncode<T>(a: T): Uint8Array {
   return new TextEncoder().encode(JSON.stringify(a));
 }
 
-export function provide<A extends Record<string, any>, B extends Record<string, any>>(
-  actions: A,
-  tasks: B = {} as B,
-) {
-  return expose<WorkerTasks & B>({
-    listenForActions: handleWorkerActions(actions),
-    ...tasks,
+export function provide<
+  C extends Record<string, any>,
+  A extends Record<string, any>,
+  T extends Record<string, any>,
+>({
+  actions,
+  connections,
+  tasks,
+}: {
+  actions?: A;
+  connections?: Record<string, PromiseWithResolvers<Comlink.Remote<C>>>;
+  tasks?: T;
+}) {
+  const allTasks = expose<WorkerTasks & T>({
+    _listen: _listen<A>(actions || ({} as A)),
+    _manage: _manage<C>(connections || {}),
+    ...(tasks || ({} as T)),
   });
+
+  return {
+    connections: connections || ({} as Record<string, PromiseWithResolvers<Comlink.Remote<C>>>),
+    tasks: allTasks,
+  };
 }
 
 export async function trackArtworkCacheId(track: Track): Promise<string> {
@@ -130,7 +156,7 @@ export function transfer<T = unknown>(a: T) {
 
 // PRIVATE
 
-function handleWorkerActions<A extends Record<string, any>>(actions: A) {
+function _listen<A extends Record<string, any>>(actions: A) {
   async function handleAction(
     port: MessagePort,
     action: {
@@ -168,3 +194,21 @@ function handleWorkerActions<A extends Record<string, any>>(actions: A) {
     };
   };
 }
+
+function _manage<C extends Record<string, any>>(
+  connections: Record<string, PromiseWithResolvers<Comlink.Remote<C>>>,
+) {
+  console.log(connections);
+
+  return (connectionId: string, workerPort: MessagePort) => {
+    let conn = connections[connectionId];
+    const remote = endpoint<C>(workerPort);
+
+    if (!conn) {
+      connections[connectionId] = Promise.withResolvers<Comlink.Remote<C>>();
+      conn = connections[connectionId];
+    }
+
+    conn.resolve(remote);
+  };
+}
diff --git a/src/scripts/configurator/input/common.ts b/src/scripts/configurator/input/common.ts
deleted file mode 100644
index 0c45efe..0000000
--- a/src/scripts/configurator/input/common.ts
+++ /dev/null
@@ -1,6 +0,0 @@
-import { CONNECTIONS } from "./constants";
-import type { Scheme } from "./types";
-
-export function isSupportedScheme(scheme: string): scheme is Scheme {
-  return !!CONNECTIONS[scheme as Scheme];
-}
diff --git a/src/scripts/configurator/input/constants.ts b/src/scripts/configurator/input/constants.ts
deleted file mode 100644
index 7bf9fcd..0000000
--- a/src/scripts/configurator/input/constants.ts
+++ /dev/null
@@ -1,5 +0,0 @@
-export const CONNECTIONS = {
-  "file+local": "/input/native-fs",
-  opensubsonic: "/input/opensubsonic",
-  s3: "/input/s3",
-};
diff --git a/src/scripts/configurator/input/types.d.ts b/src/scripts/configurator/input/types.d.ts
deleted file mode 100644
index 2a7aeee..0000000
--- a/src/scripts/configurator/input/types.d.ts
+++ /dev/null
@@ -1,3 +0,0 @@
-import type { CONNECTIONS } from "./constants";
-
-export type Scheme = keyof typeof CONNECTIONS;
diff --git a/src/scripts/configurator/input/worker.ts b/src/scripts/configurator/input/worker.ts
index 1df0eae..25675ec 100644
--- a/src/scripts/configurator/input/worker.ts
+++ b/src/scripts/configurator/input/worker.ts
@@ -1,57 +1,30 @@
 import * as URI from "uri-js";
 
-import type {
-  Consult,
-  ConsultGrouping,
-  GroupConsult,
-  InputWorkerTasks,
-  Track,
-} from "@applets/core/types";
-import type { Scheme } from "./types";
-import { CONNECTIONS } from "./constants";
-import { endpoint, groupTracksPerScheme, provide, transfer } from "@scripts/common";
-import { isSupportedScheme } from "./common";
-import type { Remote } from "comlink";
+import type { Consult, GroupConsult, InputWorkerTasks, Track } from "@applets/core/types";
+import { groupTracksPerScheme, initialConnections, provide } from "@scripts/common";
 
 ////////////////////////////////////////////
-// TASKS
+// SETUP
 ////////////////////////////////////////////
-const tasks = provide(
-  {
-    consult,
-    contextualize,
-    groupConsult,
-    list,
-    resolve,
-  },
-  {
-    manageInputWorker,
-  },
-);
+const actions = {
+  consult,
+  contextualize,
+  groupConsult,
+  list,
+  resolve,
+};
 
-export type Tasks = typeof tasks;
+const { connections, tasks } = provide({
+  actions,
+  connections: initialConnections<InputWorkerTasks>(["file+local", "opensubsonic", "s3"]),
+  tasks: actions,
+});
 
-////////////////////////////////////////////
-// CONNECTIONS
-////////////////////////////////////////////
-const connections: Record<
-  Scheme,
-  {
-    promise: Promise<Remote<InputWorkerTasks>>;
-    resolve: (value: Remote<InputWorkerTasks>) => void;
-    reject: () => void;
-  }
-> = {
-  "file+local": Promise.withResolvers(),
-  opensubsonic: Promise.withResolvers(),
-  s3: Promise.withResolvers(),
-};
+export type Actions = typeof actions;
+export type Tasks = typeof tasks;
 
-function manageInputWorker(scheme: Scheme, port: MessagePort) {
-  // TODO: Not sure what happens here when it's already resolved.
-  //       I guess ideally the connections record should always
-  //       provide the latest incoming value from here.
-  connections[scheme].resolve(endpoint<InputWorkerTasks>(port));
+function isSupportedScheme(scheme: string) {
+  return !!connections[scheme];
 }
 
 ////////////////////////////////////////////
@@ -142,7 +115,7 @@ function groupTracks(tracks: Track[]) {
   const grouped = groupTracksPerScheme(
     tracks,
     Object.fromEntries(
-      Object.entries(CONNECTIONS).map(([k, _v]) => {
+      Object.entries(connections).map(([k, _v]) => {
         return [k, []];
       }),
     ),
diff --git a/src/scripts/input/native-fs/worker.ts b/src/scripts/input/native-fs/worker.ts
index f60770e..4902152 100644
--- a/src/scripts/input/native-fs/worker.ts
+++ b/src/scripts/input/native-fs/worker.ts
@@ -9,7 +9,7 @@ import {
   recursiveList,
   trackHandleId,
 } from "./common";
-import { expose, provide, transfer } from "@scripts/common";
+import { provide, transfer } from "@scripts/common";
 
 ////////////////////////////////////////////
 // TASKS
@@ -22,7 +22,7 @@ const actions = {
   resolve,
 };
 
-const tasks = provide(actions, actions);
+const tasks = provide({ actions, tasks: actions });
 
 export type Actions = typeof actions;
 export type Tasks = typeof tasks;
diff --git a/src/scripts/input/opensubsonic/worker.ts b/src/scripts/input/opensubsonic/worker.ts
index 7d2360d..3f152bd 100644
--- a/src/scripts/input/opensubsonic/worker.ts
+++ b/src/scripts/input/opensubsonic/worker.ts
@@ -14,7 +14,7 @@ import {
   serverId,
   serversFromTracks,
 } from "./common.ts";
-import { expose, provide, transfer } from "@scripts/common.ts";
+import { provide, transfer } from "@scripts/common.ts";
 
 ////////////////////////////////////////////
 // TASKS
@@ -27,7 +27,7 @@ const actions = {
   resolve,
 };
 
-const tasks = provide(actions, actions);
+const tasks = provide({ actions, tasks: actions });
 
 export type Actions = typeof actions;
 export type Tasks = typeof tasks;
diff --git a/src/scripts/input/s3/worker.ts b/src/scripts/input/s3/worker.ts
index a1c3d90..73e9caf 100644
--- a/src/scripts/input/s3/worker.ts
+++ b/src/scripts/input/s3/worker.ts
@@ -10,7 +10,7 @@ import {
   loadBuckets,
   parseURI,
 } from "./common";
-import { expose, provide, transfer } from "@scripts/common";
+import { provide, transfer } from "@scripts/common";
 import { SCHEME } from "./constants";
 
 ////////////////////////////////////////////
@@ -24,7 +24,7 @@ const actions = {
   resolve,
 };
 
-const tasks = provide(actions, actions);
+const tasks = provide({ actions, tasks: actions });
 
 export type Actions = typeof actions;
 export type Tasks = typeof tasks;
diff --git a/src/scripts/processor/artwork/worker.ts b/src/scripts/processor/artwork/worker.ts
index cdc0557..6637b52 100644
--- a/src/scripts/processor/artwork/worker.ts
+++ b/src/scripts/processor/artwork/worker.ts
@@ -2,23 +2,27 @@ import type { IPicture } from "music-metadata";
 import * as IDB from "idb-keyval";
 
 import type { Artwork, ArtworkRequest } from "./types";
-import { expose, provide, transfer } from "@scripts/common";
+import { provide } from "@scripts/common";
 import { IDB_ARTWORK_PREFIX } from "./constants";
 import { musicMetadataTags } from "../metadata/common";
-import { getTransferables } from "@okikio/transferables";
 
 // State
 let queue: ArtworkRequest[] = [];
 
 ////////////////////////////////////////////
-// ACTIONS
+// SETUP
 ////////////////////////////////////////////
-provide({
+
+const actions = {
   artwork,
   supply,
-});
+};
+
+provide({ actions });
 
-// Actions
+////////////////////////////////////////////
+// ACTIONS
+////////////////////////////////////////////
 
 async function artwork(request: ArtworkRequest) {
   const art = await processRequest(request);
-- 
2.51.2