Fix agent websocket URL resolution and connection hang

packages/js-server-sdk/src/agent.ts

@@ -13,7 +13,7 @@ import {
   TrackEncoding,
 } from '@fishjam-cloud/fishjam-proto';
 import { AgentCallbacks, Brand, FishjamConfig, Override, PeerId } from './types';
-import { getFishjamUrl, httpToWebsocket } from './utils';
+import { getAgentWebsocketUrl } from './utils';
 
 const expectedEventsList = ['trackData'] as const;
 /**
@@ -47,32 +47,45 @@ export type AgentEvents = {
 export class FishjamAgent extends (EventEmitter as new () => TypedEmitter<AgentEvents>) {
   private readonly client: WebSocket;
 
-  private resolveConnectionPromise: ((value: void | PromiseLike<void>) => void) | null = null;
+  private resolveConnectionPromise: (() => void) | null = null;
+  private rejectConnectionPromise: ((reason: Error) => void) | null = null;
   private readonly connectionPromise: Promise<void>;
   private readonly pendingImageCaptures = new Map<string, PendingImageCapture>();
 
-  constructor(config: FishjamConfig, agentToken: string, callbacks?: AgentCallbacks) {
+  constructor(config: FishjamConfig, agentToken: string, callbacks?: AgentCallbacks, peerWebsocketUrl?: string) {
     super();
 
-    const fishjamUrl = getFishjamUrl(config);
-    const websocketUrl = `${httpToWebsocket(fishjamUrl)}/socket/agent/websocket`;
+    const websocketUrl = getAgentWebsocketUrl(config, peerWebsocketUrl);
 
     this.client = new WebSocket(websocketUrl);
 
     this.client.binaryType = 'arraybuffer';
 
     this.client.onclose = (message) => {
       this.rejectPendingCaptures('WebSocket closed');
+      // A close before the socket ever opened (e.g. a 404 on a cluster that no
+      // longer hosts the agent socket) must reject `awaitConnected`, otherwise it
+      // hangs forever.
+      this.rejectConnectionPromise?.(
+        new Error(
+          `Agent websocket closed before connecting (code ${message.code}${message.reason ? `: ${message.reason}` : ''})`
+        )
+      );
+      this.settleConnection();
       callbacks?.onClose?.(message.code, message.reason);
     };
     this.client.onerror = (message) => callbacks?.onError?.(message);
 
     this.client.onmessage = (message) => this.dispatchNotification(message);
     this.client.onopen = () => this.setupConnection(agentToken);
 
-    this.connectionPromise = new Promise<void>((resolve) => {
+    this.connectionPromise = new Promise<void>((resolve, reject) => {
       this.resolveConnectionPromise = resolve;
+      this.rejectConnectionPromise = reject;
     });
+    // Guard against an unhandled rejection when the agent is constructed directly
+    // without awaiting `awaitConnected`; callers that do await still see the error.
+    this.connectionPromise.catch(() => {});
   }
 
   /**
@@ -210,10 +223,13 @@ export class FishjamAgent extends (EventEmitter as new () => TypedEmitter<AgentE
 
     this.client.send(auth);
 
-    if (this.resolveConnectionPromise) {
-      this.resolveConnectionPromise();
-      this.resolveConnectionPromise = null;
-    }
+    this.resolveConnectionPromise?.();
+    this.settleConnection();
+  }
+
+  private settleConnection(): void {
+    this.resolveConnectionPromise = null;
+    this.rejectConnectionPromise = null;
   }
 
   private isExpectedEvent(notification: string): notification is ExpectedAgentEvents {

packages/js-server-sdk/src/client.ts

@@ -189,7 +189,7 @@ export class FishjamClient {
         peerConfig: { type: 'agent', options },
       });
 
-      const agent = new FishjamAgent(this.fishjamConfig, data.token, callbacks);
+      const agent = new FishjamAgent(this.fishjamConfig, data.token, callbacks, data.peer_websocket_url);
       await agent.awaitConnected();
 
       return { agent: agent, peer: data.peer as Peer };

packages/js-server-sdk/src/utils.ts

@@ -18,3 +18,16 @@ export const getFishjamUrl = (config: FishjamConfig) => {
     return `https://fishjam.io/api/v1/connect/${config.fishjamId}`;
   }
 };
+
+const AGENT_SOCKET_PATH = '/socket/agent/websocket';
+
+export const getAgentWebsocketUrl = (config: FishjamConfig, peerWebsocketUrl?: string): string => {
+  if (peerWebsocketUrl) {
+    // The server may return the address without a scheme (e.g. `host/socket/peer/websocket`).
+    const url = new URL(peerWebsocketUrl.includes('://') ? peerWebsocketUrl : `https://${peerWebsocketUrl}`);
+    url.protocol = url.protocol.replace('http', 'ws');
+    url.pathname = url.pathname.replace(/\/socket\/peer\/websocket$/, AGENT_SOCKET_PATH);
+    return url.href;
+  }
+  return `${httpToWebsocket(getFishjamUrl(config))}${AGENT_SOCKET_PATH}`;
+};

packages/js-server-sdk/tests/agent.test.ts

@@ -0,0 +1,100 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { getAgentWebsocketUrl } from '../src/utils';
+import { FishjamAgent } from '../src/agent';
+
+describe('getAgentWebsocketUrl', () => {
+  it('swaps the peer socket suffix for the agent one when peer_websocket_url is present', () => {
+    expect(
+      getAgentWebsocketUrl({ fishjamId: 'test-id', managementToken: 't' }, 'wss://media-node/socket/peer/websocket')
+    ).toBe('wss://media-node/socket/agent/websocket');
+  });
+
+  it('normalizes an http(s) peer_websocket_url to ws(s)', () => {
+    expect(
+      getAgentWebsocketUrl(
+        { fishjamId: 'test-id', managementToken: 't' },
+        'http://localhost:5000/socket/peer/websocket'
+      )
+    ).toBe('ws://localhost:5000/socket/agent/websocket');
+  });
+
+  it('prepends https:// when the server returns a scheme-less peer_websocket_url', () => {
+    expect(
+      getAgentWebsocketUrl(
+        { fishjamId: 'test-id', managementToken: 't' },
+        'cloud.fishjam.work/api/v1/connect/abc123/socket/peer/websocket'
+      )
+    ).toBe('wss://cloud.fishjam.work/api/v1/connect/abc123/socket/agent/websocket');
+  });
+
+  it('preserves a path prefix while swapping the suffix', () => {
+    expect(
+      getAgentWebsocketUrl(
+        { fishjamId: 'test-id', managementToken: 't' },
+        'wss://host/tenant-123/socket/peer/websocket'
+      )
+    ).toBe('wss://host/tenant-123/socket/agent/websocket');
+  });
+
+  it('falls back to deriving from a plain fishjamId when peer_websocket_url is absent', () => {
+    expect(getAgentWebsocketUrl({ fishjamId: 'test-id', managementToken: 't' })).toBe(
+      'wss://fishjam.io/api/v1/connect/test-id/socket/agent/websocket'
+    );
+  });
+
+  it('falls back to deriving from a full Fishjam URL when peer_websocket_url is absent', () => {
+    expect(
+      getAgentWebsocketUrl({ fishjamId: 'http://localhost:4000/api/v1/connect/local-id', managementToken: 't' })
+    ).toBe('ws://localhost:4000/api/v1/connect/local-id/socket/agent/websocket');
+  });
+});
+
+type MessageLike = { data: Uint8Array | ArrayBuffer };
+
+class FakeWebSocket {
+  static instances: FakeWebSocket[] = [];
+
+  binaryType = 'blob';
+  onopen: (() => void) | null = null;
+  onclose: ((event: { code: number; reason: string }) => void) | null = null;
+  onerror: ((event: unknown) => void) | null = null;
+  onmessage: ((event: MessageLike) => void) | null = null;
+  close = vi.fn();
+
+  constructor(public readonly url: string) {
+    FakeWebSocket.instances.push(this);
+  }
+
+  send() {}
+}
+
+const config = { fishjamId: 'test-id', managementToken: 'test-token' };
+
+describe('FishjamAgent connection', () => {
+  beforeEach(() => {
+    FakeWebSocket.instances = [];
+    vi.stubGlobal('WebSocket', FakeWebSocket);
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+  });
+
+  it('connects to the agent socket derived from peer_websocket_url', () => {
+    new FishjamAgent(config, 'token', undefined, 'wss://media-node/socket/peer/websocket');
+    expect(FakeWebSocket.instances.at(-1)?.url).toBe('wss://media-node/socket/agent/websocket');
+  });
+
+  it('resolves awaitConnected once the socket opens', async () => {
+    const agent = new FishjamAgent(config, 'token');
+    FakeWebSocket.instances.at(-1)?.onopen?.();
+    await expect(agent.awaitConnected()).resolves.toBeUndefined();
+  });
+
+  it('rejects awaitConnected when the socket closes before connecting', async () => {
+    const agent = new FishjamAgent(config, 'token');
+    FakeWebSocket.instances.at(-1)?.onclose?.({ code: 1006, reason: 'not found' });
+    await expect(agent.awaitConnected()).rejects.toThrow(/closed before connecting/);
+  });
+});