Experiment to rebuild Diffuse using web applets.
1import * as Uint8 from "uint8arrays";
2import * as Comlink from "comlink";
3import { xxh32 } from "xxh32";
4import { getTransferables } from "@okikio/transferables";
5
6import type { Track } from "@applets/core/types";
7import type { DiffuseApplet } from "./applet/common";
8
9// export { SharedWorkerPolyfill as SharedWorker } from "@okikio/sharedworker";
10export const SharedWorker = globalThis.SharedWorker;
11
12////////////////////////////////////////////
13// 🌳
14////////////////////////////////////////////
15
16export type WorkerTasks = {
17 _listen: ReturnType<typeof _listen>;
18 _manage: ReturnType<typeof _manage>;
19};
20
21////////////////////////////////////////////
22// 🛠️
23////////////////////////////////////////////
24
25export function arrayShuffle<T>(array: Array<T>): Array<T> {
26 if (array.length === 0) {
27 return [];
28 }
29
30 array = [...array];
31
32 for (let index = array.length - 1; index > 0; index--) {
33 const randArr = crypto.getRandomValues(new Uint32Array(1));
34 const randVal = randArr[0] / 2 ** 32;
35 const newIndex = Math.floor(randVal * (index + 1));
36 [array[index], array[newIndex]] = [array[newIndex], array[index]];
37 }
38
39 return array;
40}
41
42export function cleanUndefinedValuesForTracks(tracks: Track[]): Track[] {
43 return tracks.map((track) => {
44 const t = { ...track };
45
46 if (t.tags) {
47 if ("album" in t.tags && t.tags.album === undefined) delete t.tags.album;
48 if ("artist" in t.tags && t.tags.artist === undefined) delete t.tags.artist;
49 if ("genre" in t.tags && t.tags.genre === undefined) delete t.tags.genre;
50 if ("year" in t.tags && t.tags.year === undefined) delete t.tags.year;
51
52 if ("of" in t.tags.disc && t.tags.disc.of === undefined) delete t.tags.disc.of;
53 if ("of" in t.tags.track && t.tags.track.of === undefined) delete t.tags.track.of;
54 }
55
56 return t;
57 });
58}
59
60export function comparable(value: unknown) {
61 return xxh32(JSON.stringify(value));
62}
63
64export function endpoint<T extends Record<string, any> = WorkerTasks>(ini: Comlink.Endpoint) {
65 const e = Comlink.wrap<T>(ini);
66 if ("start" in ini && typeof ini.start === "function") ini.start();
67 return e;
68}
69
70export function expose<A extends Record<string, any>>(
71 tasks: A,
72 opts?: {
73 ports?: {
74 applets: MessagePort[];
75 consumers: MessagePort[];
76 };
77 },
78): A {
79 if (globalThis.SharedWorkerGlobalScope && self instanceof SharedWorkerGlobalScope) {
80 self.onconnect = (event: MessageEvent) => {
81 const port = event.ports[0];
82 opts?.ports?.applets?.push(port);
83 Comlink.expose(tasks, port);
84 port.start();
85 };
86
87 (self as any).connected = true;
88 } else {
89 Comlink.expose(tasks, self);
90 }
91
92 return tasks;
93}
94
95export function groupTracksPerScheme(
96 tracks: Track[],
97 initial: Record<string, Track[]> = {},
98): Record<string, Track[]> {
99 return tracks.reduce((acc: Record<string, Track[]>, track: Track) => {
100 const scheme = track.uri.split(":", 1)[0];
101 return { ...acc, [scheme]: [...(acc[scheme] || []), track] };
102 }, initial);
103}
104
105export function inIframe() {
106 return window.self !== window.top;
107}
108
109export function initialConnections<C extends Record<string, any>>(ids: string[]) {
110 const connections: Record<string, PromiseWithResolvers<Comlink.Remote<C>>> = {};
111
112 ids.forEach((c) => {
113 connections[c] = Promise.withResolvers<Comlink.Remote<C>>();
114 });
115
116 return connections;
117}
118
119export function isPrimitive(test: unknown) {
120 return test !== Object(test);
121}
122
123export function jsonDecode<T>(a: any): T {
124 return JSON.parse(new TextDecoder().decode(a));
125}
126
127export function jsonEncode<T>(a: T): Uint8Array {
128 return new TextEncoder().encode(JSON.stringify(a));
129}
130
131export function postMessages<D, T>({
132 data,
133 ports,
134 transfer,
135}: {
136 data: D;
137 ports: MessagePort[];
138 transfer?: Transferable[];
139}) {
140 ports.forEach((port) => {
141 port.postMessage(data, transfer ?? []);
142 });
143}
144
145export function provide<
146 C extends Record<string, any>,
147 A extends Record<string, any>,
148 T extends Record<string, any>,
149>({
150 actions,
151 connections,
152 tasks,
153}: {
154 actions?: A;
155 connections?: Record<string, PromiseWithResolvers<Comlink.Remote<C>>>;
156 tasks?: T;
157}) {
158 const portsHolder = {
159 applets: [] as MessagePort[],
160 consumers: [] as MessagePort[],
161 };
162
163 const allTasks = expose<WorkerTasks & T>(
164 {
165 _listen: _listen<A>(actions || ({} as A), portsHolder),
166 _manage: _manage<C>(connections || {}),
167 ...(tasks || ({} as T)),
168 },
169 {
170 ports: portsHolder,
171 },
172 );
173
174 return {
175 connections: connections || ({} as Record<string, PromiseWithResolvers<Comlink.Remote<C>>>),
176 ports: portsHolder,
177 tasks: allTasks,
178 };
179}
180
181export function sync<DataType = unknown>(context: DiffuseApplet<DataType>, port: MessagePort) {
182 port.onmessage = (event) => {
183 if (event.data?.type === "data") {
184 context.data = event.data.data;
185 }
186 };
187}
188
189export async function trackArtworkCacheId(track: Track): Promise<string> {
190 return await crypto.subtle
191 .digest("SHA-256", new TextEncoder().encode(track.uri))
192 .then((a) => Uint8.toString(new Uint8Array(a), "base64url"));
193}
194
195export function transfer<T = unknown>(a: T) {
196 const b = getTransferables(a);
197 return Comlink.transfer(a, b);
198}
199
200// PRIVATE
201
202function _listen<A extends Record<string, any>>(
203 actions: A,
204 portsHolder: {
205 applets: MessagePort[];
206 consumers: MessagePort[];
207 },
208) {
209 async function handleAction(
210 port: MessagePort,
211 action: {
212 type: "action";
213 id: string;
214 actionId: string;
215 arguments: any;
216 },
217 ) {
218 const result = await actions[action.actionId]?.(action.arguments);
219 return postMessage(port, action.id, result);
220 }
221
222 function postMessage<T>(port: MessagePort, id: string, result: T) {
223 port.postMessage(
224 {
225 type: "actioncomplete",
226 id,
227 result,
228 },
229 {
230 transfer: getTransferables(result),
231 },
232 );
233 }
234
235 return (port: MessagePort) => {
236 Comlink.expose(actions, port);
237 portsHolder.consumers.push(port);
238
239 port.onmessage = async (message) => {
240 switch (message.data?.type) {
241 case "action":
242 return handleAction(port, message.data);
243 }
244 };
245 };
246}
247
248function _manage<C extends Record<string, any>>(
249 connections: Record<string, PromiseWithResolvers<Comlink.Remote<C>>>,
250) {
251 return (connectionId: string, workerPort: MessagePort) => {
252 let conn = connections[connectionId];
253 const remote = endpoint<C>(workerPort);
254
255 if (!conn) {
256 connections[connectionId] = Promise.withResolvers<Comlink.Remote<C>>();
257 conn = connections[connectionId];
258 }
259
260 conn.resolve(remote);
261 };
262}