import { Socket, SocketOptions, ManagerOptions } from 'socket.io-client'; import { TwinTypeEnum, TwinStatusEnum, ScreenTwinReportedProperties, TwinResponse, DeviceStatus, EdgeTwinResponse, EdgeTwinDesiredPropertiesResponse, PeripheralTwinResponse, CloudTwinResponse, Instance, PeripheralInstance, TwinMessageResult, TwinMessageResultStatus, EventAckResponse, EventAckStatus, PeripheralCreateResult, PeripheralCreateOrUpdateAckResponse, PeripheralOwnershipLoss, } from './types/twin.types'; import { formatScopedHardwareId, HardwareIdScope } from './scoped-hardware-id'; import { PhyHubConnection } from './services/phyhub-connection.service'; import { PhyHubDirectConnection } from './services/phyhub-direct-connection.service'; import { CloudAppConnection, type CloudAppConnectionParams } from './services/cloud-app-connection.service'; import { WebAppConnection, type WebAppConnectionParams } from './services/web-app-connection.service'; import type { WebAppSessionError } from './services/web-app-session.service'; import { loadWebAppBoot, readStoredWebSession, readWebSessionCodeFromLocation, } from './services/web-app-session.service'; import { createWebSessionCode, WebSessionCodeIssuer, type WebSessionCode, type WebSessionCodeClientAdapter, type WebSessionCodeListener, type WebSessionCodeOptions, type WebSessionCodeSubscription, } from './services/web-session-code.service'; import { SignalsService } from './services/signals.service'; import shortenKeys from './helpers/shorten-look-ups.helper'; import { DataRequestTypeEnum, InitSignalPayload } from './types/signal.types'; import { WebRTCManager, TwinTransport, PhygridDataChannel, PhygridMediaStream, WebRTCManagerOptions, IceServersProvider, IceServersResult, MediaStreamOptions, MediaTrackKind, TurnServerConfig, MediaStreamResponderOptions, DataChannelResponderOptions, MediaStreamCallback, DataChannelCallback, PeerInfo, } from './services/webrtc'; import { TwinRegistry } from './twin-registry'; import { createTwinMessaging } from './twin-messaging'; import { DescriptorValidator } from './advisory-validation'; export interface EventPayload { twinId?: string; sourceTwinId?: string; sourceDeviceId?: string; deviceId?: string; status?: TwinStatusEnum; data?: T; } export type MyIoOptions = Partial; export type PhyHubClientParams = | { instanceId?: string; moduleName?: string; dataResidency?: string; cloudApp?: undefined; webApp?: undefined } | { cloudApp: CloudAppConnectionParams; instanceId?: undefined; moduleName?: undefined; dataResidency?: undefined; webApp?: undefined; } | { webApp: WebAppConnectionParams; instanceId?: undefined; moduleName?: undefined; dataResidency?: undefined; cloudApp?: undefined; }; export interface CloudTwinLifecycleEvent { /** UUID minted by phyhub per fan-out — caller can dedupe on this. */ eventId?: string; /** Echoes `properties.desired.$version` for ordering. */ $version?: number; twin: CloudTwinResponse; } export class PhyHubClient { private static instance: PhyHubClient | null = null; private instanceId: string | undefined = undefined; private moduleName: string | undefined = undefined; private socket: Socket | null = null; private socketConnected = false; // Idempotency guard for listener arming, keyed on socket OBJECT IDENTITY (not a // boolean): socket.io reuses one Socket instance across reconnects, so a reconnect // re-entering initializeConnection() must not stack a second handler set — while a // genuinely new socket object (fresh page, replaced connection) arms exactly once. // Without this, every disconnect/reconnect cycle appended a full extra set of // twinMessage/twinUpdated/PONG handlers: N cycles ⇒ N+1 deliveries of every twin // event, plus geometric ping keepalive growth (TECH-1334). private armedSocket: Socket | null = null; // Every handler this client armed, tracked so disconnect() can detach exactly // them — the socket itself is a page-wide singleton shared with other consumers // (window.edgeHub / window.phygridSocketSingleton) and must not be torn down. private armedSocketHandlers: Array<{ event: string; handler: (...args: any[]) => void }> = []; private armedManagerHandlers: Array<{ event: string; handler: (...args: any[]) => void }> = []; // Idempotent detach functions for per-request 'error' listeners still in // flight (armRequestErrorListener) — drained by disconnect() so abandoned // requests don't strand listeners on the shared singleton socket. private pendingRequestErrorDetachers: Set<() => void> = new Set(); private emitQueue: any[] = []; private lastDeviceStatusCheck: number = 0; private lastDeviceStatusResponse: DeviceStatus | undefined = undefined; private signals: SignalsService | undefined; private subscribedTwins: Set = new Set(); private twinMessageListeners: { [key: string]: Set<(message: any) => void> } = {}; private instances: Map = new Map(); private twinUpdateListeners: { [key: string]: Set<(twin: TwinResponse) => void> } = {}; private twinInstancesRegistry: TwinRegistry | null = null; private webrtcManager: WebRTCManager | null = null; // Cached ICE servers PER use-case (so a 'media' caller is never served a // shorter-lived 'data' credential), shared across every WebRTC connection this // client opens incl. fan-out peers. Each entry carries the epoch-ms refresh // deadline (a margin before the credential TTL). Empty/unconfigured results // (ttl 0) are negative-cached briefly so an unconfigured region doesn't cause a // per-peer round-trip storm, while still self-healing once configured. private cachedIceServers: Map<'media' | 'data', { result: IceServersResult; refreshAt: number }> = new Map(); private instanceTransport: TwinTransport | null = null; // --- cloud-app branch state --- private cloudAppParams: CloudAppConnectionParams | undefined = undefined; private cloudAppConnection: CloudAppConnection | null = null; /** Cloud twins keyed by twin.id, hydrated from `cloudAppAuthenticated` and lifecycle events. */ private cloudTwins: Map = new Map(); private cloudTwinCreatedListeners: Set<(event: CloudTwinLifecycleEvent) => void> = new Set(); private cloudTwinUpdatedListeners: Set<(event: CloudTwinLifecycleEvent) => void> = new Set(); private cloudTwinDeletedListeners: Set<(event: CloudTwinLifecycleEvent) => void> = new Set(); // --- web-app branch state --- private webAppParams: WebAppConnectionParams | undefined = undefined; private webAppConnection: WebAppConnection | null = null; private webAppListenersAttached = false; private readonly EVENTS = { PING: 'ping', PONG: 'pong', CONNECT: 'connect', RECONNECT: 'reconnect', RECONNECT_ERROR: 'reconnect_error', RECONNECT_FAILED: 'reconnect_failed', DISCONNECT: 'disconnect', REPORT_SCREEN_INSTANCE_PROPERTIES: 'setScreenInstanceReportedProperties', GET_DEVICE_STATUS: 'getDeviceStatus', GET_DEVICE_NETWORKS: 'getDeviceNetworks', GET_DEVICE_INSTANCE: 'getDeviceInstance', GET_INSTANCE: 'getInstance', SIGNAL_EVENT: 'sendEventSignal', SIGNAL_CHECKOUT: 'sendEventSignal', SIGNAL_PURCHASE: 'sendEventSignal', SIGNAL_SESSION: 'sendSessionSignal', SIGNAL_CLIENT: 'sendClientSignal', TWIN_MESSAGE: 'twinMessage', TWIN_SUBSCRIBE: 'twinSubscribe', TWIN_UNSUBSCRIBE: 'twinUnsubscribe', GET_TWIN_BY_ID: 'getTwinById', GET_ICE_SERVERS: 'getIceServers', } as const; constructor(params: PhyHubClientParams = {}) { if (params && 'cloudApp' in params && params.cloudApp) { // Cloud-app branch: server-side process, no `instanceId` / `moduleName` // / phyos-routing. The room is keyed by `appRegistrationId` derived // server-side from the JWT, so the client never has to know it. this.cloudAppParams = params.cloudApp; return; } if (params && 'webApp' in params && params.webApp) { // Web-app branch: public phone client authenticated by a QR claim code // exchanged for short-lived session tokens. No `instanceId` / // `moduleName` / phyos routing — the session identity is server-derived // from the JWT claims. this.webAppParams = params.webApp; return; } const deviceParams = params as { instanceId?: string; moduleName?: string; dataResidency?: string; }; this.instanceId = deviceParams.instanceId; if (!this.instanceId) { if (typeof window !== 'undefined') { console.log('Running in browser, looking for instanceId in URL hash', window.location.hash); this.instanceId = new URLSearchParams(window.location.hash.slice(1)).get('instanceId') || undefined; console.log('Found instanceId in URL hash', this.instanceId); } else { this.instanceId = process.env.TWIN_ID; } } this.moduleName = deviceParams.moduleName; } private async initializeConnection(): Promise { if (this.socket && this.socketConnected) { return this; } if (this.cloudAppParams) { return this.initializeCloudAppConnection(); } if (this.webAppParams) { return this.initializeWebAppConnection(); } // Web-app auto-detection: a deployed web app is a plain static page — no // device runtime, no #instanceId. Its marker is the boot.json the platform // publishes next to the bundle. Only probed in a browser with no explicit // params and no instanceId, so screen apps (instanceId in the hash) and // Node processes never pay for the fetch. if (!this.instanceId && typeof window !== 'undefined') { console.info('[web-app] no instanceId — probing ./boot.json for web-app mode'); const boot = await loadWebAppBoot(); if (boot) { console.info( `[web-app] boot.json found (urlId: ${boot.urlId}, phyhub: ${boot.phyhubUrl}, mint base: ${boot.sessionBaseUrl}) — web app mode`, ); // Boot precedence (TECH-1468): a fresh #code= wins, a stored session // resumes silently, and only with neither does the user see the // "scan the QR" error. The code/stored ordering itself (including the // expired-code fallback) lives in WebAppConnection. const code = readWebSessionCodeFromLocation(); const storedSession = readStoredWebSession(boot.urlId) ?? undefined; if (!code && !storedSession) { throw new Error( 'No session code in the URL — open this web app via its QR code or session link (#code=...).', ); } this.webAppParams = { urlId: boot.urlId, sessionBaseUrl: boot.sessionBaseUrl, phyhubUrl: boot.phyhubUrl, code, storedSession, }; return this.initializeWebAppConnection(); } console.info('[web-app] no boot.json — not a web app, continuing with the device branch'); } try { console.log( 'initializeConnection()', JSON.stringify( { instanceId: this.instanceId, moduleName: this.moduleName, }, null, 2, ), ); const phyHubConnection = PhyHubConnection.getInstance({ instanceId: this.instanceId, moduleName: this.moduleName, }); this.socket = await phyHubConnection.getPhyHubSocket(); // Wait for socket to be connected await this.waitForSocketConnected(); // Now that socket is connected, set up the listeners this.setupSocketListeners( (value: this) => value, (reason?: any) => { throw reason; }, ); return this; } catch (err) { this.socket = null; this.socketConnected = false; throw err; } } private async waitForSocketConnected(): Promise { await new Promise((resolve, reject) => { const socket = this.socket; if (socket?.connected) { this.socketConnected = true; resolve(); return; } // once() only auto-removes the listener that fired — the counterpart must // be detached explicitly, or every await-connect cycle leaves one stale // once-listener behind on the shared socket (a `connect_error` one on // every successful connect). Same accumulation class as TECH-1334. const onConnect = () => { socket?.off('connect_error', onConnectError); this.socketConnected = true; resolve(); }; const onConnectError = (error: Error) => { socket?.off('connect', onConnect); reject(error); }; socket?.once('connect', onConnect); socket?.once('connect_error', onConnectError); }); } private async initializeCloudAppConnection(): Promise { if (!this.cloudAppParams) { throw new Error('initializeCloudAppConnection(): cloudAppParams not set'); } if (!this.cloudAppConnection) { this.cloudAppConnection = new CloudAppConnection(this.cloudAppParams); } const { socket, initialTwins } = await this.cloudAppConnection.connect(); this.socket = socket; this.socketConnected = true; // Hydrate the twin map from the snapshot CloudAppConnection captured off // the `cloudAppAuthenticated` ack. Doing it synchronously here — before // this method returns — closes the race where a caller calling // `getCloudTwins()` immediately after `await connect()` would see `[]` // because the listener was attached too late. // // `initialTwins === undefined` means "no fresh snapshot" (the warm-cache // reconnect path inside CloudAppConnection); preserve whatever's in the // Map. Empty array means "server confirmed zero twins" and the existing // Map (if any) should be cleared. if (initialTwins !== undefined) { this.cloudTwins.clear(); for (const twin of initialTwins) { if (twin?.id) this.cloudTwins.set(twin.id, twin); } } // Listener arming is guarded separately from the twin-map hydration above: // hydration is data refresh and must run on EVERY (re)connect, whereas the // handlers below must be armed exactly once per socket object — cloud-app // reconnects re-enter this method with the same socket, and re-arming would // stack duplicate handler sets (TECH-1334, same defect as the device path). this.setupCloudAppSocketListeners(); return this; } private setupCloudAppSocketListeners(): void { if (!this.socket) return; if (this.armedSocket === this.socket) return; // A different socket object is being armed (fresh connection after a token // renewal or reconnect churn) — release the previous one's handler set first. this.detachArmedHandlers(); this.armedSocket = this.socket; // Re-arm: on reconnect, phyhub runs `subscribeSocketCloudAppEvents` again // and re-emits `cloudAppAuthenticated` with a fresh twin set. The // one-shot listener inside CloudAppConnection.connect() removed itself // via cleanup() after the initial event, so only this outer listener // fires on subsequent emits. this.armSocketHandler('cloudAppAuthenticated', (payload: { status?: string; twins?: CloudTwinResponse[] }) => { if (Array.isArray(payload?.twins)) { this.cloudTwins.clear(); for (const twin of payload.twins) { if (twin?.id) this.cloudTwins.set(twin.id, twin); } } }); // Phyhub fans Cloud-twin lifecycle events out as `{ eventId, $version, twin }` // (architecture-doc Phase 2 / commit bd90c0942). We normalize internally to // the `EventPayload` shape so the existing // `onTwinUpdate` / `twinUpdateListeners` machinery keeps working — at the // cost of one tiny mapping here. this.armSocketHandler('twinCreated', (payload: CloudTwinLifecycleEvent) => { this.handleCloudTwinLifecycleEvent('twinCreated', payload); }); this.armSocketHandler('twinUpdated', (payload: CloudTwinLifecycleEvent) => { this.handleCloudTwinLifecycleEvent('twinUpdated', payload); }); this.armSocketHandler('twinDeleted', (payload: CloudTwinLifecycleEvent) => { this.handleCloudTwinLifecycleEvent('twinDeleted', payload); }); // Cross-twin messaging — phyhub broadcasts to the cloud-app room with the // existing `EventPayload` shape (`twinId`/`sourceTwinId`/`data`) so this // path is identical to the device path; reuse the device twinMessage listener // bookkeeping we already have. this.armSocketHandler('twinMessage', (payload: EventPayload) => { if (payload?.twinId && this.twinMessageListeners[payload.twinId]) { this.twinMessageListeners[payload.twinId].forEach((listener) => { try { listener(payload); } catch (error) { console.error('[cloud-app] twinMessage listener threw', error); } }); } }); } private async initializeWebAppConnection(): Promise { if (!this.webAppParams) { throw new Error('initializeWebAppConnection(): webAppParams not set'); } if (!this.webAppConnection) { this.webAppConnection = new WebAppConnection(this.webAppParams); } const { socket } = await this.webAppConnection.connect(); this.socket = socket; this.socketConnected = true; // `connect()` can be re-entered via assureSocketConnection during the // routine token-expiry reconnects — attach these listeners once. if (this.webAppListenersAttached) { return this; } this.webAppListenersAttached = true; // Gate the emit queue on `webAppAuthenticated` (not bare `connect`): the // server only registers web-session handlers after the handshake ack, so // emitting earlier would drop events. socket.on('webAppAuthenticated', () => { this.socketConnected = true; this.processEmitQueue(); }); socket.on(this.EVENTS.DISCONNECT, () => { this.socketConnected = false; }); // Twin room fan-out uses the same `EventPayload` shape as // the device path, so wire it into the existing onTwinUpdate machinery. socket.on('twinUpdated', (payload: EventPayload) => { if (!payload?.data) return; const twinId = payload.twinId || payload.data.id; if (!twinId || !this.twinUpdateListeners[twinId]) return; this.twinUpdateListeners[twinId].forEach((listener) => { try { listener(payload.data!); } catch (error) { console.error(`[web-app] onTwinUpdate listener for ${twinId} threw`, error); } }); }); // Cross-twin messaging arrives with the standard `EventPayload` shape; // reuse the existing twinMessage listener bookkeeping. socket.on('twinMessage', (payload: EventPayload) => { if (payload?.twinId && this.twinMessageListeners[payload.twinId]) { this.twinMessageListeners[payload.twinId].forEach((listener) => { try { listener(payload); } catch (error) { console.error('[web-app] twinMessage listener threw', error); } }); } }); return this; } /** * Subscribe to terminal session failure (refresh grant expired/revoked or * endpoint disabled) — the app should prompt a new QR scan. Web-app branch * only. Returns an unsubscribe handle. */ public onWebAppSessionTerminated(listener: (error: WebAppSessionError) => void): () => void { if (!this.webAppParams) { throw new Error('onWebAppSessionTerminated() is only available for web app sessions'); } if (!this.webAppConnection) { this.webAppConnection = new WebAppConnection(this.webAppParams); } return this.webAppConnection.onSessionTerminated(listener); } /** * Web sessions are consumers only (getTwinById, twinSubscribe, twinMessage, * getIceServers, signals): no device identity, no reporting, no peripheral * CRUD. Guard device/cloud-only APIs with a clear error rather than a hang * or a misleading "instance not set" failure. */ private assertNotWebAppSession(operation: string): void { if (this.webAppParams) { throw new Error(`${operation} is not available for web app sessions`); } } /** * Claim-code issuing is a device-session capability: phyhub registers * `createWebSessionCode` only on authenticated device sockets (web and * cloud sessions would hang with no handler) — fail fast instead. */ private assertDeviceSession(operation: string): void { if (this.webAppParams || this.cloudAppParams) { throw new Error(`${operation} is only available for device sessions (screen/edge apps)`); } } /** * Internal — keep `cloudTwins` map fresh and fan to user listeners. * Drops the event silently if the payload doesn't carry a twin (defensive — * phyhub always emits a full twin, but a future schema bump could regress). */ private handleCloudTwinLifecycleEvent( kind: 'twinCreated' | 'twinUpdated' | 'twinDeleted', payload: CloudTwinLifecycleEvent, ): void { const twin = payload?.twin; if (!twin || !twin.id) { console.warn(`[cloud-app] ${kind} payload missing twin`, payload); return; } if (kind === 'twinDeleted') { this.cloudTwins.delete(twin.id); } else { this.cloudTwins.set(twin.id, twin); } // Fan to cloud-app-specific listeners with the original payload shape // (`{ eventId, $version, twin }`) so consumers can dedupe on eventId and // order on $version without client-side normalization. const target = kind === 'twinCreated' ? this.cloudTwinCreatedListeners : kind === 'twinUpdated' ? this.cloudTwinUpdatedListeners : this.cloudTwinDeletedListeners; target.forEach((listener) => { try { listener(payload); } catch (error) { console.error(`[cloud-app] ${kind} listener threw`, error); } }); // Also fan to the existing `onTwinUpdate` machinery (per-twin listeners // keyed by twinId) so callers don't need a separate API for cloud twins. if (kind === 'twinUpdated' && this.twinUpdateListeners[twin.id]) { this.twinUpdateListeners[twin.id].forEach((cb) => { try { cb(twin as TwinResponse); } catch (error) { console.error(`[cloud-app] onTwinUpdate listener for ${twin.id} threw`, error); } }); } } /** Register a socket handler and track it for disconnect(). */ private armSocketHandler(event: string, handler: (...args: any[]) => void): void { this.socket!.on(event, handler); this.armedSocketHandlers.push({ event, handler }); } /** Register a manager (socket.io) handler and track it for disconnect(). */ private armManagerHandler(event: string, handler: (...args: any[]) => void): void { // Manager events are typed as a closed union; the tracked-detach bookkeeping // needs the plain-string form. (this.socket!.io.on as (event: string, handler: (...args: any[]) => void) => void)(event, handler); this.armedManagerHandlers.push({ event, handler }); } /** * Arm a per-request 'error' listener on the shared socket and return an * idempotent detach function. Callers MUST invoke the detach when the * request's ack settles the promise — these listeners used to be added with * a bare `socket.on('error', ...)` and never removed, so every completed * call left one behind on the page-wide singleton (same accumulation class * as TECH-1334). If the error fires first, the listener detaches itself * before rejecting. In-flight detaches are tracked so disconnect() can * drop listeners belonging to requests that will never settle. */ private armRequestErrorListener(onSocketError: (error: any) => void): () => void { const socket = this.socket; if (!socket) { return () => {}; } const handler = (error: any) => { detach(); onSocketError(error); }; const detach = () => { this.pendingRequestErrorDetachers.delete(detach); socket.off('error', handler); }; socket.on('error', handler); this.pendingRequestErrorDetachers.add(detach); return detach; } /** * Detach every tracked handler from the socket it was armed on and reset the * tracking state. The handlers live on `armedSocket` — the socket they were * registered on — which can differ from `this.socket` when the connection * singleton has been replaced (phyhub-connection nulls it on connect_error); * detaching from `this.socket` there would no-op and leave the old socket, * possibly still alive, delivering into this client. */ private detachArmedHandlers(): void { if (this.armedSocket) { for (const { event, handler } of this.armedSocketHandlers) { this.armedSocket.off(event, handler); } for (const { event, handler } of this.armedManagerHandlers) { (this.armedSocket.io.off as (event: string, handler: (...args: any[]) => void) => void)(event, handler); } } this.armedSocketHandlers = []; this.armedManagerHandlers = []; this.armedSocket = null; } private setupSocketListeners(resolve: (value: this) => void, _reject: (reason?: any) => void): void { if (!this.socket) return; // Already armed on this exact socket object — the connection is up (the caller // awaited it), so resolve without stacking another handler set. Re-subscription // after reconnects is handled by the connect/reconnect handlers armed below on // the first pass. if (this.armedSocket === this.socket) { resolve(this); return; } // A different socket object is being armed — release the previous one's // handler set first so the abandoned socket stops delivering into this client. this.detachArmedHandlers(); this.armedSocket = this.socket; this.armSocketHandler(this.EVENTS.PONG, (data: any) => { setTimeout(() => { this.emit('ping', { count: data.data.count + 1 }); }, 30000); }); this.armSocketHandler(this.EVENTS.CONNECT, () => { this.socketConnected = true; this.emit(this.EVENTS.PING, { count: 1 }); this.processEmitQueue(); // Re-subscribe to twins after reconnection this.subscribedTwins.forEach((twinId) => { this.subscribeTwin(twinId).catch((error) => { console.error(`Failed to re-subscribe to twin ${twinId}:`, error); }); }); resolve(this); }); this.armManagerHandler(this.EVENTS.RECONNECT, () => { this.socketConnected = true; this.processEmitQueue(); // Re-subscribe to twins after reconnection this.subscribedTwins.forEach((twinId) => { this.subscribeTwin(twinId).catch((error) => { console.error(`Failed to re-subscribe to twin ${twinId}:`, error); }); }); }); this.armManagerHandler(this.EVENTS.RECONNECT_ERROR, () => { this.socketConnected = false; }); this.armManagerHandler(this.EVENTS.RECONNECT_FAILED, () => { this.socketConnected = false; }); this.armSocketHandler(this.EVENTS.DISCONNECT, () => { this.socketConnected = false; }); this.armSocketHandler( this.EVENTS.TWIN_MESSAGE, (payload: EventPayload, callback?: (response: TwinMessageResult | null) => void) => { let result: TwinMessageResult | null = null; if (payload.twinId && payload.data) { // Look up listeners by twinId (the target of the message) // This is the original pre-PR-144 approach const listeners = this.twinMessageListeners[payload.twinId]; if (listeners?.size) { listeners.forEach((listener) => { try { listener(payload); result = { status: TwinMessageResultStatus.Success, message: 'Action completed' }; } catch (error) { console.error(`[TWIN_MESSAGE] Error in listener for ${payload.twinId}:`, error); result = { status: TwinMessageResultStatus.Error, message: (error as Error)?.message || 'An error occurred', }; } }); } else { result = { status: TwinMessageResultStatus.Warning, message: `No listeners found for twin ${payload.twinId}`, }; } if (callback) { callback(result); } } }, ); // Add global listener for twin updates this.armSocketHandler('twinUpdated', (payload: EventPayload) => { console.log('twinUpdated event received', JSON.stringify(payload, null, 2)); if (payload.data) { const twinId = payload.twinId || payload.data.id; // Ownership-loss detection (TECH-1394): peripheral hardwareIds are // unique tenant-wide, so another device — or another instance on // THIS device — registering the same id takes our twin over. The // check is level-triggered against the twin's current owner identity // (deviceId + desired.instanceId), not edge-triggered on the // takeover event itself: every later twinUpdated (e.g. the new // owner's reports reaching this device's room) re-delivers the // signal, so a takeover missed during a disconnect self-heals on the // next twin activity. Only instances that ACQUIRED the twin as its // owner are considered — a consumer holding a sibling instance's // peripheral on the same device legitimately sees a foreign // instanceId forever and must not self-revoke. if (twinId && payload.data.deviceId) { const ownedPeripheral = this.twinInstancesRegistry?.getCachedPeripheralInstance(twinId); const ownDeviceId = ownedPeripheral?.edgeInstance?.deviceId; const ownInstanceId = ownedPeripheral?.edgeInstance?.id; if (ownedPeripheral?.acquiredAsOwner && ownDeviceId) { const updatedInstanceId = (payload.data.properties?.desired as { instanceId?: string } | undefined) ?.instanceId; const lostToDevice = payload.data.deviceId !== ownDeviceId; const lostToInstance = !lostToDevice && ownInstanceId !== undefined && updatedInstanceId !== undefined && updatedInstanceId !== ownInstanceId; if (lostToDevice || lostToInstance) { this.handlePeripheralOwnershipLost(twinId, { newOwnerDeviceId: lostToDevice ? payload.data.deviceId : undefined, newOwnerInstanceId: lostToInstance ? updatedInstanceId : undefined, }); } } } if (twinId && this.twinUpdateListeners[twinId]) { // Invoke all callbacks registered for this twin this.twinUpdateListeners[twinId].forEach((cb) => { try { cb(payload.data!); console.log( 'twinUpdateListeners[twinId].forEach() callback executed', JSON.stringify(payload.data, null, 2), ); } catch (error) { console.error(`Error in twin update listener for ${twinId}:`, error); } }); } } }); } /** * Revoke a peripheral this app lost to another device or to another * instance on this device (TECH-1394): * - mark the pooled PeripheralTwinInstance ownership-lost (its * emit/updateReported start rejecting locally, its transport gate stops * delivering incoming actions, onOwnershipLost fires so the app * releases the hardware); * - drop it from the registry pool, so a later re-acquire (after taking * the twin back via createPeripheralTwin) builds a fresh instance; * - unsubscribe from the twin room, server-side AND locally — removing * the id from `subscribedTwins` is what stops the reconnect * re-subscribe from silently rejoining the room we were evicted from. * (For a same-device loss the room barely matters — sibling instances * share the device room broadcast — which is why the transport gate, * not the unsubscribe, is the effective cutoff there.) */ private handlePeripheralOwnershipLost(twinId: string, loss: PeripheralOwnershipLoss = {}): void { const peripheralInstance = this.twinInstancesRegistry?.getCachedPeripheralInstance(twinId); if (!peripheralInstance) { return; } const newOwnerLabel = loss.newOwnerDeviceId ? ` by device ${loss.newOwnerDeviceId}` : loss.newOwnerInstanceId ? ` by instance ${loss.newOwnerInstanceId} on this device` : ''; console.warn( `handlePeripheralOwnershipLost(): peripheral ${twinId} was taken over${newOwnerLabel} — revoking the local instance`, ); this.subscribedTwins.delete(twinId); this.twinInstancesRegistry?.removePeripheralInstance(twinId); // Best-effort server-side unsubscribe: phyhub already evicted our // sockets from the twin room at takeover; this covers the case where // that eviction failed or raced our reconnect. try { const payload: EventPayload = { twinId, sourceTwinId: peripheralInstance.edgeInstance?.id }; this.emit(this.EVENTS.TWIN_UNSUBSCRIBE, payload, () => undefined); } catch (error) { console.warn(`Failed to send twinUnsubscribe for taken-over peripheral ${twinId}`, error); } peripheralInstance.markOwnershipLost(loss); } private async assureSocketConnection(): Promise { if (!this.socket || !this.socketConnected) { await this.initializeConnection(); } if (!this.socket) { throw new Error('Socket not initialized'); } } // Single-flights connect(): concurrent callers (React StrictMode mounts an // effect twice in dev) must all await the SAME initialization — returning // `instance` while initializeConnection is still in flight hands out a // client whose branch params (webAppParams/cloudAppParams) are not set yet, // so branch-gated APIs like onWebAppSessionTerminated() throw spuriously. private static connectInFlight: Promise | null = null; public static async connect(params: PhyHubClientParams = {}): Promise { console.info(`connect(): Connecting to phyhub`); // TODO handle deviceId and accessKey for screen instance scenarios if (!PhyHubClient.connectInFlight) { PhyHubClient.connectInFlight = (async () => { PhyHubClient.instance = new PhyHubClient(params); await PhyHubClient.instance.initializeConnection(); // The TwinRegistry is a peripheral-twin instance pool — Cloud apps and // web sessions don't own peripheral twins, so we skip it for those branches. if (!PhyHubClient.instance.cloudAppParams && !PhyHubClient.instance.webAppParams) { PhyHubClient.instance.twinInstancesRegistry = new TwinRegistry(PhyHubClient.instance); } console.info(`connect(): Connection to phyhub initialized`); return PhyHubClient.instance; })(); // A failed initialization must not poison every later connect() with a // half-built instance — reset so the next call retries from scratch // (e.g. a web app reopened with a fresh #code after a rejected one). PhyHubClient.connectInFlight.catch(() => { PhyHubClient.connectInFlight = null; PhyHubClient.instance = null; }); } return PhyHubClient.connectInFlight; } public getSocket = (): Socket | null => { return this.socket; }; public initializeSignals = async (initParams?: InitSignalPayload) => { let edgeTwin; if (!initParams) { // Web sessions have no device status/edge twin to derive the payload // from — the app must pass its own InitSignalPayload explicitly. if (this.webAppParams) { throw new Error('initializeSignals() requires explicit initParams for web app sessions'); } await this.getDeviceStatus(); if (!this.lastDeviceStatusResponse) { throw new Error('Failed to fetch device settings'); } edgeTwin = (await this.getInstance()) as Instance; if (!edgeTwin) { throw new Error('Unable to determine app settings'); } } const initSignalsPayload: InitSignalPayload = { deviceId: initParams?.deviceId ?? edgeTwin?.deviceId ?? this.lastDeviceStatusResponse?.deviceId ?? '', installationId: initParams?.installationId ?? edgeTwin?.properties.desired.installationId ?? '', spaceId: initParams?.spaceId ?? this.lastDeviceStatusResponse?.spaceId ?? '', tenantId: initParams?.tenantId ?? this.lastDeviceStatusResponse?.tenantId ?? '', appVersion: initParams?.appVersion ?? 'XXXXXXXXXXXXXXXXXXXXXXXX', appId: initParams?.appId ?? 'XXXXXXXXXXXXXXXXXXXXXXXX', environment: initParams?.environment ?? this.lastDeviceStatusResponse?.gridEnv ?? '', dataResidency: (initParams?.dataResidency ?? this.lastDeviceStatusResponse?.dataResidency ?? '').toUpperCase(), country: initParams?.country ?? 'SE', installationVersion: initParams?.installationVersion ?? 'XXXXXXXXXXXXXXXXXXXXXXXX', accessToken: initParams?.accessToken ?? this.lastDeviceStatusResponse?.accessKey, clientUserAgent: initParams?.clientUserAgent ?? undefined, ip: initParams?.ip ?? this.lastDeviceStatusResponse?.ip?.[0]?.ipv4 ?? 'N/A', }; this.signals = new SignalsService(this, initSignalsPayload); return this.signals; }; public isConnected(): boolean { return this.socketConnected; } public async sendSignal(type: DataRequestTypeEnum, data: Record) { // The EVENTS.SIGNAL_* constants are the phyos agent's method names // (`sendEventSignal`, ...). phyhub's direct-connection sockets (cloud/web // sessions) accept the same spelling as an alias, so one name works on // every transport — no per-transport translation here. let eventToEmit = undefined; switch (type) { case DataRequestTypeEnum.EVENT: eventToEmit = this.EVENTS.SIGNAL_EVENT; break; case DataRequestTypeEnum.CHECKOUT: eventToEmit = this.EVENTS.SIGNAL_CHECKOUT; break; case DataRequestTypeEnum.PURCHASE: eventToEmit = this.EVENTS.SIGNAL_PURCHASE; break; case DataRequestTypeEnum.SESSION: eventToEmit = this.EVENTS.SIGNAL_SESSION; break; case DataRequestTypeEnum.CLIENT: eventToEmit = this.EVENTS.SIGNAL_CLIENT; break; default: throw new Error(`Unsupported type ${type}`); } if (!eventToEmit) { throw new Error(`Unsupported type ${type}`); } const payload: EventPayload> = { data, }; this.emit(eventToEmit, payload); } /** * Resolve the device's current spaceId for space-scoped hardwareId minting * (TECH-1394). Bypasses getDeviceStatus's 10s cache window: a started-but- * unanswered status request would serve `undefined` from the cache slot, * failing a healthy boot-time mint (connect → initializeSignals → * generateScopedHardwareId races the signals-triggered status request). * Serves the cached status when it already carries a spaceId; otherwise * asks directly with its own timeout. */ private async resolveDeviceSpaceId(timeoutMs = 3000): Promise { const cachedSpaceId = this.lastDeviceStatusResponse?.spaceId; if (cachedSpaceId) { return cachedSpaceId; } await this.assureSocketConnection(); return new Promise((resolve, reject) => { const timeoutId = setTimeout(() => { detachErrorListener(); reject( new Error( `generateScopedHardwareId: could not get spaceId — getDeviceStatus gave no answer within ${timeoutMs}ms`, ), ); }, timeoutMs); const detachErrorListener = this.armRequestErrorListener((error: any) => { clearTimeout(timeoutId); reject(error); }); this.emit(this.EVENTS.GET_DEVICE_STATUS, (response: any) => { clearTimeout(timeoutId); detachErrorListener(); const spaceId = response?.spaceId; if (!spaceId) { reject(new Error('generateScopedHardwareId: the device status response did not include a spaceId')); return; } resolve(spaceId); }); }); } public async getDeviceStatus(): Promise { this.assertNotWebAppSession('getDeviceStatus()'); await this.assureSocketConnection(); const now = Date.now(); const tenSeconds = 10 * 1000; if (now - this.lastDeviceStatusCheck < tenSeconds) { // console.info('getDeviceStatus(): Skipping emit - returning last known status'); return Promise.resolve(this.lastDeviceStatusResponse); } return new Promise((resolve, reject) => { this.lastDeviceStatusCheck = now; const detachErrorListener = this.armRequestErrorListener(reject); this.emit(this.EVENTS.GET_DEVICE_STATUS, (response: any) => { detachErrorListener(); this.lastDeviceStatusResponse = response; resolve(response); }); }); } /** * Get a DataChannel connection to a target twin. * Initiates a WebRTC DataChannel connection for peer-to-peer messaging. * @param targetTwinId - The twin ID to connect to * @param channelName - Optional channel name for multiple channels to same peer (default: 'default') */ public getDataChannel = async (targetTwinId: string, channelName?: string): Promise => { const manager = await this.getWebRTCManager(); return manager.createDataChannel(targetTwinId, channelName); }; /** * Arm a data-channel responder on this instance: many initiators, each as its * own peer connection and data channel. The callback fires ONCE PER CONNECTED * PEER with that peer's own channel and peer id (N=1 is the single-peer case). * @returns a stop() function that tears down the responder and all peers. */ public onDataChannel = async ( callback: DataChannelCallback, options?: DataChannelResponderOptions, ): Promise<() => void> => { const manager = await this.getWebRTCManager(); // Subscribe on this instance's own twin — that's where remote initiators // address their offers; each peer's answers route back to its offer source. return manager.acceptDataChannel(this.instanceTransport!.twinId, callback, options); }; /** * Get a MediaStream connection to a target twin. * Initiates a WebRTC MediaStream connection for peer-to-peer media. * @param targetTwinId - The twin ID to connect to * @param options - Optional MediaStream options (including channelName for multiple streams) */ public async getMediaStream( targetTwinId: string, options?: MediaStreamOptions, ): Promise<{ stream: PhygridMediaStream; close: () => void; }> { const manager = await this.getWebRTCManager(); const channelName = options?.channelName ?? 'default'; const stream = await manager.createMediaStream(targetTwinId, options, channelName); return { stream, close: () => stream.close(), }; } /** * Arm a media responder on this instance: one local source fanned out to many * simultaneous peers/viewers, each as its own peer connection. The callback * fires ONCE PER CONNECTED PEER with that peer's own stream and peer id (N=1 is * the single-viewer case). * @returns a stop() function that tears down the responder and all peers. */ public onMediaStream = async ( callback: MediaStreamCallback, options?: MediaStreamResponderOptions, ): Promise<() => void> => { const manager = await this.getWebRTCManager(); return manager.acceptMediaStream(this.instanceTransport!.twinId, options ?? {}, callback); }; /** * Fetch ICE servers (STUN + short-lived TURN credentials) over the authenticated * PhyHub socket. The device/tenant identity is taken server-side from the socket, * so no deviceId is sent. Used as the default `iceServersProvider` for * getWebRTCManager() — the credentials never have to be wired in by the caller. * * Rejects (rather than hangs) if no `getIceServers` handler answers within the * timeout — e.g. an older phyhub without the handler. The PeerConnectionManager * treats a throw as "fall back to static STUN", so this stays backward-compatible. */ public async getIceServers(useCase?: 'media' | 'data'): Promise { const cacheKey: 'media' | 'data' = useCase === 'media' ? 'media' : 'data'; // Reuse cached credentials while within their refresh window so repeated // connections — and fan-out (N peers from one source) — don't each round-trip // phyhub. Keyed by use-case so a 'media' caller is never served a 'data' TTL. const cached = this.cachedIceServers.get(cacheKey); if (cached && Date.now() < cached.refreshAt) { return cached.result; } const payload: EventPayload<{ useCase?: 'media' | 'data' }> = { data: useCase ? { useCase } : {}, }; // Short negative-cache window for empty/unconfigured (ttl 0) results, so an // unconfigured region collapses a fan-out's N calls to one round-trip while // still self-healing within seconds once the region is configured. const emptyResultCacheMs = 30 * 1000; const requestTimeoutMs = 10 * 1000; const result = await new Promise((resolve, reject) => { let timeoutId: ReturnType | undefined; // Remove the one-off 'error' listener (and timer) on every exit path — // getIceServers is the per-connection default provider, so a leaked listener // per call would accumulate on the long-lived socket. const finish = (): void => { if (timeoutId) clearTimeout(timeoutId); this.socket?.off('error', onError); }; const onError = (error: unknown): void => { finish(); reject(error instanceof Error ? error : new Error(`getIceServers(): socket error ${String(error)}`)); }; // The timeout bounds the WHOLE operation — including a stalled (re)connect: // assureSocketConnection runs inside it so a hung socket can't block forever. timeoutId = setTimeout(() => { finish(); reject(new Error(`getIceServers(): request timed out after ${requestTimeoutMs}ms`)); }, requestTimeoutMs); void (async () => { try { await this.assureSocketConnection(); this.socket?.on('error', onError); this.emit(this.EVENTS.GET_ICE_SERVERS, payload, (response: IceServersResult) => { finish(); if (!response || !Array.isArray(response.iceServers)) { reject(new Error('getIceServers(): no iceServers in response')); return; } resolve(response); }); } catch (connectError) { finish(); reject(connectError instanceof Error ? connectError : new Error(`getIceServers(): ${String(connectError)}`)); } })(); }); // Credential-bearing results refresh a margin before expiry; empty/unconfigured // results get the short negative-cache window above. const refreshAt = result.ttlSeconds && result.ttlSeconds > 0 ? Date.now() + (result.ttlSeconds * 1000 - Math.min(30 * 1000, result.ttlSeconds * 1000 * 0.5)) : Date.now() + emptyResultCacheMs; this.cachedIceServers.set(cacheKey, { result, refreshAt }); return result; } /** * Get the WebRTCManager for advanced control over WebRTC connections. * Provides access to events, connection state, and more. */ public async getWebRTCManager(options?: WebRTCManagerOptions): Promise { if (!this.webrtcManager) { await this.assureSocketConnection(); if (!this.instanceTransport) { await this.getInstance(); // ensures transport is created } // Default the ICE servers provider to the authenticated-socket fetch so every // WebRTC connection automatically gets TURN credentials. A caller-supplied // provider always wins. On a throw (older phyhub, transient failure) the // PeerConnectionManager falls back to static STUN — today's behaviour. const resolvedOptions: WebRTCManagerOptions = options ? Object.assign({}, options) : {}; if (!resolvedOptions.iceServersProvider) { // Request the 'media' (longer, 1h-floor) TTL by default: one shared // provider serves both data channels and long-lived media relays, and a // generous TTL safely covers both — a data channel with a 1h credential is // harmless, whereas a media call on a 5-min credential would drop mid-stream. resolvedOptions.iceServersProvider = () => this.getIceServers('media'); } this.webrtcManager = new WebRTCManager(this.instanceTransport!, resolvedOptions); } else if (options) { // The manager is created once and cached, so options passed on a later call // cannot reconfigure it. Warn rather than silently ignore — especially since // the default now injects a TURN-credential provider, a dropped override // (e.g. a custom iceServersProvider) would silently use the wrong ICE source. console.warn( '[PhyHubClient] getWebRTCManager(options) called after the manager was already created; ' + 'options are ignored. Pass options on the first getWebRTCManager() call.', ); } return this.webrtcManager; } // TODO properties should automatically be updated when the edge twin is updated // We could do a refresh method and call that on reconnect like we handle the resubscribes for peripherals public async getInstance(): Promise { this.assertNotWebAppSession('getInstance()'); // TODO support multiple instances of different twin ids if (!this.instanceId) { throw new Error('Instance ID not set'); } if (this.instances.has(this.instanceId)) { return this.instances.get(this.instanceId)!; } await this.assureSocketConnection(); let instance: Instance | undefined; let instanceTwin = await this.getTwinById(this.instanceId); // Create transport once, shared by messaging and WebRTC const transport: TwinTransport = { sendMessage: async (targetTwinId, data) => { this.sendEvent(targetTwinId, data); }, subscribe: async (twinId) => { await this.subscribeTwin(twinId); }, onMessage: (twinId, callback) => { this.onTwinMessage(twinId, callback); }, offMessage: (twinId, callback) => { this.offTwinMessage(twinId, callback); }, twinId: instanceTwin.id, }; this.instanceTransport = transport; // Create messaging methods using the shared factory const messaging = createTwinMessaging(transport, 'edgeInstance'); const getPeripheralTwins = async (): Promise => { const payload: EventPayload = { data: { instanceId: instanceTwin.id, }, }; return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener(reject); this.emit('getPeripheralTwins', payload, (response: EventAckResponse) => { detachErrorListener(); // Resolving [] on an error ack would make a failed lookup // indistinguishable from "no peripherals exist", which drives // find-or-create logic into duplicate registrations. if (response?.status !== EventAckStatus.Success) { reject(new Error(response?.message || 'Failed to get peripheral twins')); return; } resolve((response.twins ?? []) as PeripheralTwinResponse[]); }); }); }; const createPeripheralTwinDetailed = async ( peripheralName: string, hardwareId: string, desiredProperties?: Record, descriptors?: Record, ): Promise => { // No client-side "already exists" guard: the server's create-or-update // (keyed TENANT-WIDE by tenant+hardwareId, TECH-1394) is the uniqueness // authority, and re-calling create IS the supported way to // re-register — refresh the name, re-point the owner device and // instance, update descriptor pins. Requires a phyhub with the // peripheral upsert (TECH-408). const payload: EventPayload = { data: { deviceId: instanceTwin.deviceId, tenantId: instanceTwin.tenantId, instanceId: instanceTwin.id, peripheralName, hardwareId, desiredProperties, descriptors, }, }; return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener(reject); this.emit('createPeripheralTwin', payload, (response: PeripheralCreateOrUpdateAckResponse) => { detachErrorListener(); // An error ack carries the rejection reason in `message` // (validation failure, unresolved descriptor) — surface it; // throwing here would escape the socket callback as an // uncaughtException and leave this promise pending forever. if (response?.status !== EventAckStatus.Success || !response.twin) { reject(new Error(response?.message || 'Failed to create peripheral twin')); return; } // A handover means this registration took the twin from another // device/instance (TECH-1394). Expected ONCE after an app move; // repeatedly seeing it means two deployments are fighting over the // same hardwareId — surface it even when the app ignores the // detailed result. if (response.handover) { console.warn( `createPeripheralTwin(): took over peripheral ${response.twin.id} (hardwareId ${hardwareId})` + `${response.handover.previousDeviceId ? ` from device ${response.handover.previousDeviceId}` : ''}` + `${response.handover.previousInstanceId ? ` (previous instance ${response.handover.previousInstanceId})` : ''}`, ); } resolve({ twin: response.twin, // Older phyhubs do not send `created`; treat absence as unknown- // but-existing (false) rather than inventing a create. created: response.created === true, handover: response.handover, }); }); }); }; const createPeripheralTwin = async ( peripheralName: string, hardwareId: string, desiredProperties?: Record, descriptors?: Record, ): Promise => { const result = await createPeripheralTwinDetailed(peripheralName, hardwareId, desiredProperties, descriptors); return result.twin as PeripheralTwinResponse; }; const generateScopedHardwareId = async (userId: string, scope: HardwareIdScope = 'global'): Promise => { if (scope === 'device') { return formatScopedHardwareId(userId, scope, instanceTwin.deviceId); } if (scope === 'space') { return formatScopedHardwareId(userId, scope, await this.resolveDeviceSpaceId()); } return formatScopedHardwareId(userId, scope); }; const getDataChannel = async (targetTwinId: string, channelName?: string) => { return await this.getDataChannel(targetTwinId, channelName); }; const onDataChannel = async ( callback: DataChannelCallback, options?: DataChannelResponderOptions, ): Promise<() => void> => { return await this.onDataChannel(callback, options); }; const getMediaStream = async (targetTwinId: string, options?: MediaStreamOptions) => { return await this.getMediaStream(targetTwinId, options); }; const onMediaStream = async ( callback: MediaStreamCallback, options?: MediaStreamResponderOptions, ): Promise<() => void> => { return await this.onMediaStream(callback, options); }; const updateReported = async (properties: Record) => { const newReported = { ...properties, }; const result = await this.updateReportedProperties(instanceTwin.id, newReported, instanceTwin.type); Object.assign(instance!, result); return result; }; instance = { ...instanceTwin, emit: messaging.emit, on: messaging.on, off: messaging.off, to: messaging.to, createPeripheralTwin, createPeripheralTwinDetailed, generateScopedHardwareId, updateReported, getPeripheralTwins, getDataChannel, onDataChannel, getMediaStream, onMediaStream, }; this.instances.set(this.instanceId, instance); return instance; } public async getEdgeInstance(): Promise { return this.getInstance(); } public async getScreenInstance(): Promise { return this.getInstance(); } // TODO properties should automatically be updated when the edge twin is updated /** * provides instance of PeripheralTwinInstance class */ public async getPeripheralInstance( peripheralTwinId: string, advisoryValidator?: DescriptorValidator, ): Promise { this.assertNotWebAppSession('getPeripheralInstance()'); if (!this.twinInstancesRegistry) { throw new Error('PhyHubClient instance is not initialized yet'); } // advisoryValidator is the opt-in for advisory descriptor validation; omitting // it leaves byte-identical existing behavior. Apps may instead call // peripheralInstance.enableAdvisoryValidation(...) after acquiring the instance. const peripheralInstance = await this.twinInstancesRegistry.getPeripheralInstance( peripheralTwinId, advisoryValidator, ); return peripheralInstance; } private getGridApp() { let currentWindow: Window & typeof globalThis = window; while (currentWindow) { try { if ((currentWindow as any).gridapp && typeof (currentWindow as any).gridapp.getSettings === 'function') { return (currentWindow as any).gridapp; } if (currentWindow.parent === currentWindow) break; currentWindow = currentWindow.parent as Window & typeof globalThis; } catch { break; } } return null; } public async getSettings(): Promise> { if (this.webAppParams) { // Web sessions get their resolved settings from the endpoint's Web // twin, delivered on the `webAppAuthenticated` ack (spec 5.4) — no // extra round trip. await this.assureSocketConnection(); const webTwin = this.webAppConnection?.getWebTwin(); if (!webTwin) { throw new Error('getSettings(): web session has no web twin yet (degraded auth ack)'); } const settings = webTwin.properties?.desired?.settings; return (settings ?? {}) as Partial; } if (typeof window !== 'undefined') { const gridapp = this.getGridApp(); if (gridapp) { return gridapp.getSettings() as Partial; } } // Existing fallback logic const edgeInstance = await this.getInstance(); if (!edgeInstance) { throw new Error('Edge instance not found'); } return (edgeInstance.properties.desired as EdgeTwinDesiredPropertiesResponse).settings; } public async getEdgeSettings(): Promise> { return this.getSettings(); } public async getScreenSettings(): Promise> { return this.getSettings(); } // todo: fix return type, remove any public async getDeviceInstance(): Promise { this.assertNotWebAppSession('getDeviceInstance()'); await this.assureSocketConnection(); return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener(reject); this.emit(this.EVENTS.GET_DEVICE_INSTANCE, (response: any) => { detachErrorListener(); resolve(response); }); }); } // todo: fix return type, remove any public async getDeviceNetworks(): Promise { this.assertNotWebAppSession('getDeviceNetworks()'); await this.assureSocketConnection(); return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener(reject); this.emit(this.EVENTS.GET_DEVICE_NETWORKS, (response: any) => { detachErrorListener(); resolve(response); }); }); } // todo: fix return type, remove any public async setScreenInstanceReportedProperties(payload: EventPayload): Promise { this.assertNotWebAppSession('setScreenInstanceReportedProperties()'); await this.assureSocketConnection(); console.info(`setScreenInstanceReportedProperties(): payload ${JSON.stringify(payload)}`); return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener(reject); this.emit(this.EVENTS.REPORT_SCREEN_INSTANCE_PROPERTIES, payload, (response: any) => { detachErrorListener(); resolve(response); }); }); } // todo: fix return type, remove any for clarity public emit(method: string, ...args: any[]): void { if (!this.socket) { console.error('emit(): Socket not created'); return; } const callback = typeof args[args.length - 1] === 'function' ? args.pop() : undefined; let emitArgs: any[]; // In direct connection mode, emit directly to the event name // In normal mode, emit to instanceId channel with method in payload (for phyos routing) if (PhyHubDirectConnection.isEnabled() || this.webAppParams) { // Direct mode / web-app sessions: emit to event name directly — web // sockets talk straight to phyhub, no phyos instanceId routing wrapper. const payload = args[0] || {}; emitArgs = [method, payload]; } else { // Normal mode: emit to instanceId channel with method in payload const event = this.instanceId; const payload = { method, ...args[0] }; emitArgs = [event, payload]; } if (callback) { emitArgs.push(callback); } if (!this.socketConnected) { console.info('emit(): Socket not connected, adding to emitQueue', emitArgs); this.emitQueue.push(emitArgs); return; } // console.info('hub-client is connected, emit(): Emitting', emitArgs); // TODO: fix ts-ignore // @ts-ignore this.socket.emit(...emitArgs); } // todo: fix return type, remove any private processEmitQueue(): void { if (!this.socket || !this.socketConnected) { return; } while (this.emitQueue.length > 0) { const args = this.emitQueue.shift(); if (args) { this.socket.emit.apply(this.socket, args); } } } public onTwinUpdate(targetTwinId: string, callback: (twin: TwinResponse) => void) { console.log(`onTwinUpdate() called for twin ${targetTwinId}`); this.assureSocketConnection(); // Initialize the set of listeners for this twin if it doesn't exist if (!this.twinUpdateListeners[targetTwinId]) { this.twinUpdateListeners[targetTwinId] = new Set(); } // Add the callback to the set of listeners this.twinUpdateListeners[targetTwinId].add(callback); console.log( `Added twin update listener for ${targetTwinId}. Total listeners: ${this.twinUpdateListeners[targetTwinId].size}`, ); // Make sure we're subscribed to this twin if (!this.subscribedTwins.has(targetTwinId)) { this.subscribeTwin(targetTwinId).catch((error) => { console.error(`Failed to subscribe to twin ${targetTwinId} for updates:`, error); }); } } public offTwinUpdate(targetTwinId: string, callback: (twin: TwinResponse) => void): boolean { console.log(`Removing update listener for twin ${targetTwinId}`); if (!this.twinUpdateListeners[targetTwinId]) { console.log(`No update listeners found for twin ${targetTwinId}`); return false; } const result = this.twinUpdateListeners[targetTwinId].delete(callback); if (this.twinUpdateListeners[targetTwinId].size === 0) { delete this.twinUpdateListeners[targetTwinId]; console.log(`Removed last update listener for twin ${targetTwinId}, cleaning up`); } else { console.log(`Removed update listener. Remaining listeners: ${this.twinUpdateListeners[targetTwinId].size}`); } return result; } /** * Send a fire-and-forget event to a twin. * This is the standard way to send events - no response is expected. * * Cloud-app branch: when `cloudAppParams` is set we route to phyhub's * `signalEvent` channel (the cloud-app authorization gate). The legacy * device path's source-twin/source-device wrapping is irrelevant — phyhub * derives `appRegistrationId` from the JWT and asserts the twin belongs to * the cloud-app's Gridapp, not from the payload's sourceTwinId. */ public sendEvent(targetTwinId: string, data: any): void { if (this.cloudAppParams) { this.sendCloudAppEvent(data); return; } if (this.webAppParams) { // phyhub stamps the server-trusted identities (`web-session:{sessionId}` // and the session's Web twin) over these fields and defaults an absent // sourceTwinId — sending it when known just keeps the wire explicit. const webPayload: EventPayload = { twinId: targetTwinId, data, }; const webTwin = this.webAppConnection?.getWebTwin(); if (webTwin) { webPayload.sourceTwinId = webTwin.id; } this.emit(this.EVENTS.TWIN_MESSAGE, webPayload); return; } // For fire-and-forget, we need to get the instance synchronously if possible // or queue the message. We'll use a simpler approach here. const instanceId = this.instanceId; if (!instanceId) { console.error('[sendEvent] Instance ID not set, cannot send event'); return; } // Get deviceId from cached instance if available const cachedInstance = this.instances.get(instanceId); const deviceId = cachedInstance?.deviceId || this.lastDeviceStatusResponse?.deviceId; if (!deviceId) { console.error('[sendEvent] Device ID not available, cannot send event'); return; } const payload: EventPayload = { twinId: targetTwinId, sourceTwinId: instanceId, sourceDeviceId: deviceId, data, }; // Fire-and-forget: just emit, no callback, no promise this.emit(this.EVENTS.TWIN_MESSAGE, payload); } /** * Cloud-app sendEvent path. The phyhub authorization gate * (subscribeSocketCloudAppEvents → SIGNAL_EVENT → authorizeCloudAppSignal) * checks that `data.deviceId` resolves to a Cloud twin owned by this cloud-app's * Gridapp, and that `data.spaceId` belongs to the same tenant. * * Caller must include `deviceId` and `spaceId` in the data payload — these * come from the connected Cloud twin (see `getTwins()` / `getTwinByDeviceId()`). * * `tenantId` is NOT taken from the payload: phyhub stamps it from the twin * that `deviceId` resolves to (TECH-1372). Sending a different one has no * effect beyond a server-side warning. */ private sendCloudAppEvent(data: any): void { if (!this.socket) { console.error('[cloud-app sendEvent] socket not connected'); return; } this.socket.emit('signalEvent', { data: shortenKeys(data) }); } /** Get all Cloud twins (cloud-app branch only). */ public getCloudTwins(): CloudTwinResponse[] { return Array.from(this.cloudTwins.values()); } /** * Subscribe to Cloud twin lifecycle events. Returns an unsubscribe handle. * Cloud-app branch only — fires for events delivered to the * `cloud-app-${appRegistrationId}` room. */ public onCloudTwinCreated(listener: (event: CloudTwinLifecycleEvent) => void): () => void { this.cloudTwinCreatedListeners.add(listener); return () => this.cloudTwinCreatedListeners.delete(listener); } public onCloudTwinUpdated(listener: (event: CloudTwinLifecycleEvent) => void): () => void { this.cloudTwinUpdatedListeners.add(listener); return () => this.cloudTwinUpdatedListeners.delete(listener); } public onCloudTwinDeleted(listener: (event: CloudTwinLifecycleEvent) => void): () => void { this.cloudTwinDeletedListeners.add(listener); return () => this.cloudTwinDeletedListeners.delete(listener); } /** * Release this client's connection. * * Cloud-app branch: stops the JWT refresh timer and disconnects the socket * (the socket is owned by this client, so tearing it down is safe). * * Device/screen branch: detaches every handler this client armed on the * shared socket and clears the listener registries. The socket itself is a * page-wide singleton (window.edgeHub / window.phygridSocketSingleton) shared * with other consumers, so it is left connected — only this client's handlers * are removed. WebRTC connections are NOT torn down here; use the stop() * handles returned by onDataChannel/onMediaStream/getMediaStream. * * The client stays reusable: the next operation re-initializes the connection * and re-arms listeners. */ public disconnect(): void { if (this.cloudAppConnection) { this.cloudAppConnection.disconnect(); this.cloudAppConnection = null; this.cloudTwins.clear(); } if (this.webAppConnection) { this.webAppConnection.disconnect(); this.webAppConnection = null; this.webAppListenersAttached = false; } // Detach from armedSocket (not this.socket): they are the same object in the // normal case, but if the connection singleton was replaced without a re-arm, // the handlers still live on the old socket. this.detachArmedHandlers(); // Requests still in flight will never settle for this client — drop their // per-request 'error' listeners too (each detach captured its own socket, // and removes itself from the set — hence the copy). for (const detachRequestErrorListener of Array.from(this.pendingRequestErrorDetachers)) { detachRequestErrorListener(); } this.socket = null; this.socketConnected = false; this.emitQueue = []; this.subscribedTwins.clear(); this.twinMessageListeners = {}; this.twinUpdateListeners = {}; // Cached instances hold messaging whose request/response listeners lived in // the registries cleared above — drop them so the next getInstance() / // getPeripheralInstance() rebuilds working ones. this.instances.clear(); if (this.twinInstancesRegistry) { this.twinInstancesRegistry = new TwinRegistry(this); } } /** * @deprecated Use instance.request() or instance.to(twinId).request() instead. * This method uses socket.io acknowledgements which may not work in all scenarios. * The instance API uses a more reliable emit/on pattern for request-response. */ public async request( targetTwinId: string, data: any, callback?: (response: TwinMessageResult) => void, ): Promise { console.log('[request] Sending to:', targetTwinId, 'data type:', data?.type || 'unknown'); const edgeInstance = await this.getInstance(); if (!edgeInstance) { console.error('[request] Edge instance not found'); throw new Error('Edge instance not found'); } const { id: edgeTwinId, deviceId } = edgeInstance; if (!deviceId) { console.error('[request] Device ID not available'); throw new Error('Device ID not available - ensure device is connected'); } const payload: EventPayload = { twinId: targetTwinId, sourceTwinId: edgeTwinId, sourceDeviceId: deviceId, data, }; if (callback) { // Callback pattern: use socket.io acknowledgement this.emit(this.EVENTS.TWIN_MESSAGE, payload, (response: TwinMessageResult) => { callback(response); }); return; } // Promise pattern: wait for response with timeout return new Promise((resolve, reject) => { const ACTION_TIMEOUT = 10 * 1000; const timeoutId = setTimeout(() => { detachErrorListener(); if (this.socket) { this.socket?.off(payload.data.type); } reject({ status: TwinMessageResultStatus.Error, message: `Request timed out after ${ACTION_TIMEOUT}ms`, }); }, ACTION_TIMEOUT); const detachErrorListener = this.armRequestErrorListener((error: any) => { clearTimeout(timeoutId); console.error('[request] Socket error:', error); reject(error); }); if (this.socket) { this.socket.on(payload.data.type, (response: TwinMessageResult) => { if (response && response.status !== TwinMessageResultStatus.Warning) { clearTimeout(timeoutId); detachErrorListener(); this.socket?.off(payload.data.type); resolve(response); } }); } this.emit(this.EVENTS.TWIN_MESSAGE, payload, (response: TwinMessageResult) => { // Socket.io acknowledgement received clearTimeout(timeoutId); detachErrorListener(); if (this.socket) { this.socket?.off(payload.data.type); } resolve(response); }); }); } public async onTwinMessage(targetTwinId: string, callback: (message: any) => void): Promise { if (!this.twinMessageListeners[targetTwinId]) { this.twinMessageListeners[targetTwinId] = new Set(); } this.twinMessageListeners[targetTwinId].add(callback); } public offTwinMessage(targetTwinId: string, callback: (message: any) => void): boolean { console.log(`Removing message listener for twin ${targetTwinId}`); if (!this.twinMessageListeners[targetTwinId]) { console.log(`No listeners found for twin ${targetTwinId}`); return false; } const result = this.twinMessageListeners[targetTwinId].delete(callback); if (this.twinMessageListeners[targetTwinId].size === 0) { delete this.twinMessageListeners[targetTwinId]; console.log(`Removed last listener for twin ${targetTwinId}, cleaning up`); } else { console.log(`Removed listener. Remaining listeners: ${this.twinMessageListeners[targetTwinId].size}`); } return result; } public removeMessageListener(twinId: string, callback: (message: any) => void): void { const listeners = this.twinMessageListeners[twinId]; if (listeners) { listeners.delete(callback); if (listeners.size === 0) { delete this.twinMessageListeners[twinId]; } } } public async subscribeTwin(targetTwinId: string, callback?: (response: any) => void) { this.subscribedTwins.add(targetTwinId); if (this.webAppParams) { await this.assureSocketConnection(); if (!this.webAppConnection) { throw new Error('subscribeTwin(): web app connection not initialized'); } // The registry lives in WebAppConnection so subscriptions are replayed // after every routine token-expiry reconnect — `twin-{id}` room // membership does not survive a disconnect. this.webAppConnection.registerSubscription(targetTwinId); if (callback) callback(undefined); return; } const edgeInstance = await this.getInstance(); if (!edgeInstance) { throw new Error('Edge instance not found'); } const { id: edgeTwinId, deviceId } = edgeInstance; const payload: EventPayload = { twinId: targetTwinId, sourceTwinId: edgeTwinId, sourceDeviceId: deviceId, }; if (callback) { this.emit(this.EVENTS.TWIN_SUBSCRIBE, payload, (response: any) => { callback(response); }); } else { return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener((error: any) => { console.error(`[subscribeTwin] Socket error:`, error); reject(error); }); this.emit(this.EVENTS.TWIN_SUBSCRIBE, payload, (response: any) => { detachErrorListener(); resolve(response); }); }); } } /** * Stop receiving events for a twin subscribed via subscribeTwin(). * * Web sessions: drops it from the reconnect-replay registry (so the routine * token-expiry reconnects stop re-joining the room) and leaves the * `twin-{id}` room server-side. Device/screen/cloud: emits twinUnsubscribe * on the socket and removes the twin from this client's re-subscribe set — * same wire call the peripheral takeover path uses. */ public unsubscribeTwin(targetTwinId: string): void { this.subscribedTwins.delete(targetTwinId); if (this.webAppParams) { this.webAppConnection?.unregisterSubscription(targetTwinId); return; } if (this.socket && this.socketConnected) { try { this.emit(this.EVENTS.TWIN_UNSUBSCRIBE, { twinId: targetTwinId }, () => undefined); } catch (error) { console.warn(`Failed to send twinUnsubscribe for twin ${targetTwinId}`, error); } } } public async getTwinById(twinId: string): Promise { console.log(`[getTwinById] Fetching twin: ${twinId}`); const payload: EventPayload<{ twinId: string }> = { data: { twinId }, }; return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener((error: any) => { console.error(`[getTwinById] Socket error for ${twinId}:`, error); reject(error); }); this.emit(this.EVENTS.GET_TWIN_BY_ID, payload, (response: any) => { detachErrorListener(); console.log(`[getTwinById] Response for ${twinId}:`, response); const { twin } = response; if (!twin) { reject(new Error(`Twin with id ${twinId} not found`)); return; } resolve(twin); }); }); } /** * The slice of this client the web-session-code service needs, kept as an * adapter so the issuing logic stays independently testable. */ private buildWebSessionCodeClientAdapter(): WebSessionCodeClientAdapter { return { ensureConnection: () => this.assureSocketConnection(), emit: (method, payload, ack) => this.emit(method, payload, ack), getTwinById: (twinId) => this.getTwinById(twinId), getSocket: () => this.socket, }; } /** * Mint ONE claim code for a Web endpoint (device sessions only). The * returned `url` is the complete public link (`{root}/{urlId}/#code={code}`) * to render as a QR. Identify the endpoint by exactly one of: * - `twinId` — the Web twin's `_id`, as stored by a settings twin picker * - `endpointId` — the Web twin's `deviceId`, as shown in Console/CLI * * Codes expire after the endpoint's `codeTtlSeconds`; for a continuously * displayed QR use subscribeWebSessionCode() instead, which rotates them. */ public async createWebSessionCode(options: WebSessionCodeOptions): Promise { this.assertDeviceSession('createWebSessionCode()'); return createWebSessionCode(this.buildWebSessionCodeClientAdapter(), options); } /** * Keep a fresh claim code available for a Web endpoint (device sessions * only): mints the initial code, rotates before expiry, re-mints after * reconnects, and retries failures with capped backoff. The listener * receives the initial code and every update through the same callback: * * ```ts * const subscription = client.subscribeWebSessionCode( * { twinId: settings.webRemote.id }, * (state) => { * // state.status === 'active' → render state.url as a QR * // state.status === 'unavailable' → hide the QR (state.retryAt says when * // the next attempt fires; null = gave up) * }, * ); * subscription.stop(); * ``` * * Runtime failures never throw — they arrive as `unavailable` states. Only * programmer errors throw synchronously (bad target, non-device session). */ public subscribeWebSessionCode( options: WebSessionCodeOptions, listener: WebSessionCodeListener, ): WebSessionCodeSubscription { this.assertDeviceSession('subscribeWebSessionCode()'); const issuer = new WebSessionCodeIssuer(this.buildWebSessionCodeClientAdapter(), options, listener); issuer.start(); return issuer; } // private async updateTwinById(twinId: string, newTwin: Partial): Promise { // const payload: EventPayload = { // twinId, // data: newTwin // }; // return new Promise((resolve, reject) => { // this.emit('updateTwin', payload, (response: any) => { // const { twin } = response; // if (!twin) { // reject(new Error(`Failed to update twin ${twinId}`)); // return; // } // resolve(twin); // }); // this.socket?.on('error', (error: any) => { // reject(error); // }); // }); // } public async updateReportedProperties( twinId: string, reportedProperties: Record, twinType?: TwinTypeEnum, ): Promise { // Web sessions are consumers — no reporting APIs (spec 5.4). this.assertNotWebAppSession('updateReportedProperties()'); // Cloud-app branch: emit on the dedicated `reportCloudTwinProperties` // wire event (phyhub's cloud-app subscriber listens on REPORT_CLOUD). // Emit directly on the cloud-app socket — no instanceId routing wrapper // since cloud-app sockets don't go through phyos. if (this.cloudAppParams) { if (!this.socket) { throw new Error('Cloud-app socket not connected'); } const payload: EventPayload = { twinId, data: reportedProperties }; return new Promise((resolve, reject) => { this.socket!.emit( 'reportCloudTwinProperties', payload, (response: { status?: string; message?: string; twin?: TwinResponse }) => { if (!response?.twin) { reject(new Error(response?.message || `Failed to update reported properties for cloud twin ${twinId}`)); return; } resolve(response.twin); }, ); }); } const twinTypeToEvent: Record = { [TwinTypeEnum.Screen]: 'reportScreenTwinProperties', [TwinTypeEnum.Edge]: 'reportEdgeTwinProperties', [TwinTypeEnum.Peripheral]: 'reportPeripheralTwinProperties', }; const eventName = twinTypeToEvent[twinType as string]; if (!eventName) { throw new Error(`Cannot update reported properties: unsupported twin type "${twinType}"`); } const payload: EventPayload = { twinId, data: reportedProperties, }; return new Promise((resolve, reject) => { const detachErrorListener = this.armRequestErrorListener(reject); this.emit(eventName, payload, (response: EventAckResponse) => { detachErrorListener(); // The error ack's `message` carries the server's reason (e.g. the // peripheral ownership guard) — same pattern as the cloud branch. if (response?.status !== EventAckStatus.Success || !response.twin) { // A typed ownership rejection is the takeover signal for clients // that missed the twinUpdated notification (offline during the // takeover, reconnect race) — trigger the same revocation so the // app still gets onOwnershipLost instead of erroring forever // (TECH-1394). if (response?.errorType === 'PeripheralOwnershipError') { this.handlePeripheralOwnershipLost(twinId); } reject(new Error(response?.message || `Failed to update reported properties for twin ${twinId}`)); return; } resolve(response.twin); }); }); } } export const connectPhyClient = (params: PhyHubClientParams = {}) => PhyHubClient.connect(params); export type { EdgeTwinResponse, Instance, PeripheralInstance, TwinMessageResult, InitSignalPayload, CloudTwinResponse }; /** @deprecated Use PeripheralInstance instead */ export type { PeripheralInstance as IPeripheralTwinInstance }; export { TwinMessageResultStatus }; export { formatScopedHardwareId, HARDWARE_ID_USER_PART_MAX_LENGTH } from './scoped-hardware-id'; export type { HardwareIdScope } from './scoped-hardware-id'; export type { PeripheralCreateResult, PeripheralHandover, PeripheralCreateOrUpdateAckResponse, PeripheralOwnershipLoss, } from './types/twin.types'; export { CloudAppConnection } from './services/cloud-app-connection.service'; export type { CloudAppConnectionParams } from './services/cloud-app-connection.service'; export { exchangeCloudAppToken, CloudAppTokenExchangeError } from './services/cloud-app-token.service'; export type { CloudAppToken, CloudAppTokenExchangeParams } from './services/cloud-app-token.service'; export { WebAppConnection } from './services/web-app-connection.service'; export type { WebAppConnectionParams, WebAppConnectResult } from './services/web-app-connection.service'; export { exchangeWebAppSession, WebAppSessionError, readWebSessionCodeFromLocation, clearWebSessionCodeFromLocation, loadWebAppBoot, readStoredWebSession, writeStoredWebSession, clearStoredWebSession, WEB_SESSION_STORAGE_KEY_PREFIX, } from './services/web-app-session.service'; export type { WebAppBoot } from './services/web-app-session.service'; export type { WebAppSession, WebAppSessionExchangeParams, WebAppSessionFetch, StoredWebSession, WebSessionStorageLike, } from './services/web-app-session.service'; export { WebSessionCodeIssuer, WebSessionCodeConfigError } from './services/web-session-code.service'; export type { WebSessionCode, WebSessionCodeOptions, WebSessionCodeState, WebSessionCodeListener, WebSessionCodeSubscription, WebSessionCodeTarget, } from './services/web-session-code.service'; export default { connectPhyClient, }; // Re-export WebRTC types for convenience export type { PhygridDataChannel, PhygridMediaStream, WebRTCManagerOptions, IceServersProvider, IceServersResult, MediaStreamOptions, MediaTrackKind, TurnServerConfig, MediaStreamResponderOptions, DataChannelResponderOptions, MediaStreamCallback, DataChannelCallback, PeerInfo, }; export { WebRTCManager, ensureWebRTCGlobals } from './services/webrtc'; // One-shot HTTP-signaled answer (WHEP) surface export type { SignalingSink, SignalingMessageType } from './services/webrtc'; export { NoMediaResponderError, PeerCapReachedError, MissingMediaTrackError } from './services/webrtc'; // Re-export twin messaging types for custom implementations export type { TwinMessagingMethods } from './twin-messaging'; export { createTwinMessaging } from './twin-messaging'; // Opt-in advisory descriptor-validation API (build-time-embed approach). export type { DescriptorValidator, CompiledValidator, AdvisoryLogger, AdvisoryFacet } from './advisory-validation'; export { validateEmittedFacet } from './advisory-validation';