client: add worker/fork support to web client

packages/client/src/index.ts

@@ -1,19 +1,17 @@
-import { ConnectRPCObjectOptions, MediaObjectCreateOptions, ScryptedStatic } from "@scrypted/types";
+import { ConnectRPCObjectOptions, ForkOptions, ForkWorker, MediaObjectCreateOptions, PluginFork, ScryptedInterface, ScryptedInterfaceProperty, ScryptedStatic } from "@scrypted/types";
 import * as eio from 'engine.io-client';
-import { SocketOptions } from 'engine.io-client';
 import { timeoutPromise } from "../../../common/src/promise-utils";
 import type { ClusterObject, ConnectRPCObject } from '../../../server/src/cluster/connect-rpc-object';
+import { domFetch } from "../../../server/src/fetch";
+import { httpFetch } from '../../../server/src/fetch/http-fetch';
 import type { IOSocket } from '../../../server/src/io';
 import { MediaObject } from '../../../server/src/plugin/mediaobject';
+import { PluginAPIProxy, PluginRemote } from "../../../server/src/plugin/plugin-api";
 import { attachPluginRemote } from '../../../server/src/plugin/plugin-remote';
 import { RpcPeer } from '../../../server/src/rpc';
 import { createRpcDuplexSerializer, createRpcSerializer } from '../../../server/src/rpc-serializer';
 import packageJson from '../package.json';
 import { isIPAddress } from "./ip";
-
-import { domFetch } from "../../../server/src/fetch";
-import { httpFetch } from '../../../server/src/fetch/http-fetch';
-
 export * as rpc from '../../../server/src/rpc';
 export * as rpc_serializer from '../../../server/src/rpc-serializer';
 
@@ -33,6 +31,15 @@ const sourcePeerId = RpcPeer.generateId();
 
 type IOClientSocket = eio.Socket & IOSocket;
 
+interface InternalFork extends Pick<ScryptedClientStatic, 'loginResult' | 'username' | 'address' | 'connectionType'> {
+    extraHeaders: {
+        [header: string]: string,
+    };
+    transports?: string[] | undefined;
+    clientName?: string;
+    admin: boolean;
+};
+
 function once(socket: IOClientSocket, event: 'open' | 'message') {
     return new Promise<any[]>((resolve, reject) => {
         const err = (e: any) => {
@@ -70,6 +77,7 @@ export interface ScryptedClientStatic extends ScryptedStatic {
     connectionType: ScryptedClientConnectionType;
     rpcPeer: RpcPeer;
     loginResult: ScryptedClientLoginResult;
+    fork<T>(options: ForkOptions & { worker: Worker }): PluginFork<T>;
 }
 
 export interface ScryptedConnectionOptions {
@@ -428,11 +436,10 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
     const eioPath = `endpoint/${pluginId}/engine.io/api`;
     const eioEndpoint = baseUrl ? new URL(eioPath, baseUrl).pathname : '/' + eioPath;
     // https://github.com/socketio/engine.io/issues/690
-    const cacheBust = Math.random().toString(36).substring(3, 10);
-    const eioOptions: Partial<SocketOptions> = {
+    const eioOptions: eio.SocketOptions = {
         path: eioEndpoint,
         query: {
-            cacheBust,
+            cacheBust: Math.random().toString(36).substring(3, 10),
         },
         withCredentials: true,
         extraHeaders,
@@ -470,7 +477,7 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
         tryLocalAddressess: tryAddresses,
     });
 
-    const localEioOptions: Partial<SocketOptions> = {
+    const localEioOptions: eio.SocketOptions = {
         ...eioOptions,
         extraHeaders: {
             ...eioOptions.extraHeaders,
@@ -586,6 +593,8 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
             endpointManager,
             mediaManager,
             clusterManager,
+            pluginHostAPI,
+            pluginRemoteAPI,
         } = scrypted;
         console.log('api attached', Date.now() - start);
 
@@ -612,6 +621,168 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
             .map(id => systemManager.getDeviceById(id))
             .find(device => device.pluginId === '@scrypted/core' && device.nativeId === `user:${username}`);
 
+        const connectRPCObject = clusterSetup(address, connectionType, queryToken, extraHeaders, options?.transports, sourcePeerId, clientName);
+
+        const loginResult: ScryptedClientLoginResult = {
+            username,
+            token,
+            directAddress,
+            localAddresses,
+            externalAddresses,
+            scryptedCloud,
+            queryToken,
+            authorization,
+            cloudAddress,
+            hostname,
+            serverId,
+        };
+
+        type ForkType = ScryptedClientStatic['fork'];
+        const fork: ForkType = (forkOptions) => {
+            const { worker } = forkOptions;
+
+            const serializer = createRpcSerializer({
+                sendMessageBuffer: buffer => worker.postMessage(buffer),
+                sendMessageFinish: message => worker.postMessage(JSON.stringify(message)),
+            });
+
+            const threadPeer = new RpcPeer("main-client", 'thread', (message, reject, serializationContext) => {
+                try {
+                    serializer.sendMessage(message, reject, serializationContext);
+                }
+                catch (e) {
+                    reject?.(e as Error);
+                }
+            });
+
+            rpcPeer.killed.finally(() => threadPeer.kill('main rpc peer killed'));
+
+            worker.addEventListener('message', async event => {
+                if (event.data instanceof Uint8Array) {
+                    serializer.onMessageBuffer(Buffer.from(event.data));
+                }
+                else {
+                    serializer.onMessageFinish(JSON.parse(event.data));
+                }
+            });
+
+            serializer.setupRpcPeer(threadPeer);
+
+            // there is no worker close event?
+            const forkApi = new PluginAPIProxy(pluginHostAPI, mediaManager);
+            threadPeer.killed.finally(() => {
+                forkApi.removeListeners();
+                worker.terminate();
+            });
+
+            const internalFork: InternalFork = {
+                loginResult,
+                username,
+                address,
+                connectionType,
+                extraHeaders,
+                transports: options?.transports,
+                clientName,
+                admin,
+            };
+
+            threadPeer.params['client'] = internalFork;
+
+            const result = (async () => {
+                const getRemote = await threadPeer.getParam('getRemote');
+                const remote = await getRemote(forkApi, pluginId, {
+                    serverVersion
+                }) as PluginRemote;
+
+                await remote.setSystemState(systemManager.getSystemState());
+                forkApi.listen((id, eventDetails, eventData) => {
+                    // ScryptedDevice events will be handled specially and repropagated by the remote.
+                    if (eventDetails.eventInterface === ScryptedInterface.ScryptedDevice) {
+                        if (eventDetails.property === ScryptedInterfaceProperty.id) {
+                            // a change on the id property means device was deleted
+                            remote.updateDeviceState(eventData, undefined);
+                        }
+                        else {
+                            // a change on anything else is a descriptor update
+                            remote.updateDeviceState(id, systemManager.getSystemState()[id]);
+                        }
+                        return;
+                    }
+
+                    if (eventDetails.property && !eventDetails.mixinId) {
+                        remote.notify(id, eventDetails, systemManager.getSystemState()[id]?.[eventDetails.property]).catch(() => { });
+                    }
+                    else {
+                        remote.notify(id, eventDetails, eventData).catch(() => { });
+                    }
+                });
+
+                const fork = await threadPeer.getParam('fork');
+                return fork;
+            })();
+
+            result.catch(() => {
+                threadPeer.kill('fork setup failed');
+                worker.terminate();
+            });
+
+            return {
+                [Symbol.dispose]() {
+                    worker.terminate();
+                    threadPeer.kill('disposed');
+                },
+                result,
+                worker: {
+                    terminate() {
+                        worker.terminate();
+                    },
+                    nativeWorker: worker,
+                } as any as ForkWorker,
+            };
+        };
+
+        const ret: ScryptedClientStatic = {
+            userId: userDevice?.id,
+            serverVersion,
+            username,
+            pluginRemoteAPI,
+            address,
+            connectionType,
+            admin,
+            systemManager,
+            clusterManager,
+            deviceManager,
+            endpointManager,
+            mediaManager,
+            disconnect() {
+                rpcPeer.kill('disconnect requested');
+            },
+            pluginHostAPI,
+            rpcPeer,
+            loginResult,
+            connectRPCObject,
+            fork,
+            connect: undefined,
+        }
+
+        socket.on('close', () => {
+            rpcPeer.kill('socket closed');
+        });
+
+        rpcPeer.killed.finally(() => {
+            socket.close();
+            ret.onClose?.();
+        });
+
+        return ret;
+    }
+    catch (e) {
+        socket.close();
+        throw e;
+    }
+}
+
+function clusterSetup(address: string, connectionType: ScryptedClientConnectionType, queryToken: any, extraHeaders: { [header: string]: string }, transports: string[] | undefined, sourcePeerId: string, clientName?: string) {
     const clusterPeers = new Map<number, Promise<RpcPeer>>();
     const finalizationRegistry = new FinalizationRegistry((clusterPeer: RpcPeer) => {
         clusterPeer.kill('object finalized');
@@ -628,17 +799,17 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
             const eioPath = 'engine.io/connectRPCObject';
             const eioEndpoint = new URL(eioPath, address).pathname;
             const eioQueryToken = connectionType === 'http' ? undefined : queryToken;
-                const clusterPeerOptions = {
+            const clusterPeerOptions: eio.SocketOptions = {
                 path: eioEndpoint,
                 query: {
-                        cacheBust,
+                    cacheBust: Math.random().toString(36).substring(3, 10),
                     clusterObject: JSON.stringify(clusterObject),
                     ...eioQueryToken,
                 },
                 withCredentials: true,
                 extraHeaders,
                 rejectUnauthorized: false,
-                    transports: options?.transports,
+                transports,
             };
 
             const clusterPeerSocket = new eio.Socket(address, clusterPeerOptions);
@@ -802,11 +973,87 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
         }
     }
 
+    return connectRPCObject;
+}
+
+export async function connectScryptedClientFork(forkMain: (client: ScryptedClientStatic) => Promise<any>) {
+    const start = Date.now();
+
+    try {
+
+        const serializer = createRpcSerializer({
+            sendMessageBuffer: buffer => self.postMessage(buffer),
+            sendMessageFinish: message => self.postMessage(JSON.stringify(message)),
+        });
+
+        const rpcPeer = new RpcPeer('thread', "main-client", (message, reject, serializationContext) => {
+            try {
+                serializer.sendMessage(message, reject, serializationContext);
+            }
+            catch (e) {
+                reject?.(e as Error);
+            }
+        });
+
+        self.addEventListener('message', event => {
+            if (event.data instanceof Uint8Array) {
+                serializer.onMessageBuffer(Buffer.from(event.data));
+            }
+            else {
+                serializer.onMessageFinish(JSON.parse(event.data));
+            }
+        });
+
+        serializer.setupRpcPeer(rpcPeer);
+
+
+        const scrypted = await attachPluginRemote(rpcPeer, undefined);
+        const {
+            serverVersion,
+            systemManager,
+            deviceManager,
+            endpointManager,
+            mediaManager,
+            clusterManager,
+            pluginHostAPI,
+            pluginRemoteAPI,
+        } = scrypted;
+        console.log('api attached', Date.now() - start);
+
+        mediaManager.createMediaObject = async<T extends MediaObjectCreateOptions>(data: any, mimeType: string, options: T) => {
+            return new MediaObject(mimeType, data, options) as any;
+        }
+        console.log('api initialized', Date.now() - start);
+
+        const {
+            loginResult,
+            username,
+            address,
+            connectionType,
+            extraHeaders,
+            transports,
+            clientName,
+            admin,
+        } = await rpcPeer.getParam('client') as InternalFork;
+
+        const { queryToken } = loginResult;
+
+        const userDevice = Object.keys(systemManager.getSystemState())
+            .map(id => systemManager.getDeviceById(id))
+            .find(device => device.pluginId === '@scrypted/core' && device.nativeId === `user:${username}`);
+
+        const connectRPCObject = clusterSetup(address, connectionType, queryToken, extraHeaders, transports, sourcePeerId, clientName);
+
+        type ForkType = ScryptedClientStatic['fork'];
+        const fork: ForkType = (forkOptions) => {
+            throw new Error('not implemented');
+        };
+
         const ret: ScryptedClientStatic = {
             userId: userDevice?.id,
             serverVersion,
             username,
-            pluginRemoteAPI: undefined,
+            pluginRemoteAPI,
             address,
             connectionType,
             admin,
@@ -818,39 +1065,24 @@ export async function connectScryptedClient(options: ScryptedClientOptions): Pro
             disconnect() {
                 rpcPeer.kill('disconnect requested');
             },
-            pluginHostAPI: undefined,
+            pluginHostAPI,
             rpcPeer,
-            loginResult: {
-                username,
-                token,
-                directAddress,
-                localAddresses,
-                externalAddresses,
-                scryptedCloud,
-                queryToken,
-                authorization,
-                cloudAddress,
-                hostname,
-                serverId,
-            },
+            loginResult,
             connectRPCObject,
-            fork: undefined,
+            fork,
             connect: undefined,
         }
-
-        socket.on('close', () => {
-            rpcPeer.kill('socket closed');
-        });
-
         rpcPeer.killed.finally(() => {
-            socket.close();
+            self.close();
             ret.onClose?.();
         });
 
-        return ret;
+        const forked = await forkMain(ret);
+
+        rpcPeer.params['fork'] = forked;
     }
     catch (e) {
-        socket.close();
+        self.close();
         throw e;
     }
 }