feat(sim): use SFU handshake endpoints in single_server_sfu mode

Backend/public/mobile-sim.js

@@ -138,9 +138,20 @@ const API = {
   },
 
   ops: {
+    ready: () => API.request('/ops/ready'),
     listRecordings: () => API.request('/recordings/me/list'),
     getRecordingDownloadUrl: (recordingId) => API.request(`/recordings/${recordingId}/download-url`),
     listNotifications: () => API.request('/push-notifications/me'),
+  },
+
+  sfu: {
+    getSession: (id) => API.request(`/streams/${id}/sfu/session`),
+    createPublishTransport: (id) => API.request(`/streams/${id}/sfu/publish-transport`, { method: 'POST', body: JSON.stringify({ role: 'camera' }) }),
+    connectPublishTransport: (id, payload) => API.request(`/streams/${id}/sfu/publish-transport/connect`, { method: 'POST', body: JSON.stringify(payload) }),
+    createSubscribeTransport: (id) => API.request(`/streams/${id}/sfu/subscribe-transport`, { method: 'POST', body: JSON.stringify({ role: 'viewer' }) }),
+    connectSubscribeTransport: (id, payload) => API.request(`/streams/${id}/sfu/subscribe-transport/connect`, { method: 'POST', body: JSON.stringify(payload) }),
+    produce: (id, payload) => API.request(`/streams/${id}/sfu/produce`, { method: 'POST', body: JSON.stringify(payload) }),
+    consume: (id, payload) => API.request(`/streams/${id}/sfu/consume`, { method: 'POST', body: JSON.stringify(payload) }),
   }
 };
 
@@ -166,10 +177,26 @@ let webrtcConnected = false;
 let hasWebrtcEverConnected = false;
 let lastPeerConnectionState = null;
 let pendingRemoteCandidates = [];
+let mediaMode = 'legacy';
 const rtcConfig = {
   iceServers: [{ urls: 'stun:stun.l.google.com:19302' }],
 };
 
+const isSfuMode = () => mediaMode === 'single_server_sfu';
+
+const detectMediaMode = async () => {
+  try {
+    const ready = await API.ops.ready();
+    const mode = ready?.checks?.mediaMode;
+    if (mode === 'single_server_sfu' || mode === 'legacy') {
+      mediaMode = mode;
+      addActivity('Media', `Mode: ${mediaMode}`);
+    }
+  } catch {
+    mediaMode = 'legacy';
+  }
+};
+
 const init = async () => {
   // Load local storage
   const saved = localStorage.getItem('mobileSimDevice');
@@ -184,6 +211,7 @@ const init = async () => {
     const session = await API.auth.getSession();
     if (session && session.session) {
       store.update({ session });
+      await detectMediaMode();
       if (store.get().deviceToken) {
         // If we have a token, skip onboarding
         navigateBasedOnRole();
@@ -660,6 +688,82 @@ const startOfferToClient = async (streamSessionId, requesterDeviceId) => {
   });
 };
 
+const startSfuPublishHandshake = async (streamSessionId) => {
+  const ready = await startCameraPreview();
+  if (!ready || !localCameraStream) {
+    throw new Error('Camera stream unavailable for SFU publish');
+  }
+
+  const publishTransport = await API.sfu.createPublishTransport(streamSessionId);
+  const transportId = publishTransport?.transport?.transportId;
+  if (!transportId) {
+    throw new Error('Missing SFU publish transport id');
+  }
+
+  await API.sfu.connectPublishTransport(streamSessionId, {
+    transportId,
+    dtlsParameters: { role: 'auto' },
+  });
+
+  const track = localCameraStream.getVideoTracks()[0];
+  await API.sfu.produce(streamSessionId, {
+    transportId,
+    kind: 'video',
+    rtpParameters: {
+      codec: 'VP8',
+      source: 'mobile-sim-camera',
+      trackSettings: track?.getSettings?.() ?? {},
+    },
+  });
+
+  addActivity('SFU', 'Publish handshake completed');
+};
+
+const setClientSfuConnectedState = () => {
+  if (remoteStreamWaitTimer) {
+    clearTimeout(remoteStreamWaitTimer);
+    remoteStreamWaitTimer = null;
+  }
+  setClientStreamPlaceholderText('Connected via SFU (simulated media path)');
+  setClientStreamMode('connecting');
+};
+
+const startSfuSubscribeHandshake = async (streamSessionId) => {
+  const subscribeTransport = await API.sfu.createSubscribeTransport(streamSessionId);
+  const transportId = subscribeTransport?.transport?.transportId;
+  if (!transportId) {
+    throw new Error('Missing SFU subscribe transport id');
+  }
+
+  await API.sfu.connectSubscribeTransport(streamSessionId, {
+    transportId,
+    dtlsParameters: { role: 'auto' },
+  });
+
+  let consumeResult = null;
+  for (let attempt = 0; attempt < 8; attempt += 1) {
+    try {
+      consumeResult = await API.sfu.consume(streamSessionId, {
+        transportId,
+        rtpCapabilities: { browser: 'mobile-sim' },
+      });
+      if (consumeResult?.consumer?.consumerId) {
+        break;
+      }
+    } catch {
+      // Camera may still be establishing producer; retry briefly.
+    }
+    await sleep(350);
+  }
+
+  if (!consumeResult?.consumer?.consumerId) {
+    throw new Error('SFU consumer was not created');
+  }
+
+  setClientSfuConnectedState();
+  addActivity('SFU', `Subscribed to producer ${consumeResult.consumer.producerId}`);
+};
+
 const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
 
 const finalizeRecordingForStream = async (streamSessionId, captureResult) => {
@@ -738,6 +842,7 @@ const connectSocket = () => {
   socket.on('connect', () => {
     store.update({ socketConnected: true });
     addActivity('System', 'Connected to realtime server');
+    void detectMediaMode();
     if (store.get().device?.role === 'camera') {
       startCameraPreview();
     }
@@ -765,7 +870,9 @@ const connectSocket = () => {
         await API.streams.accept(streamId);
         await API.streams.getPublishCreds(streamId);
         await startLocalRecording();
-        if (payload.sourceDeviceId) {
+        if (isSfuMode()) {
+          await startSfuPublishHandshake(streamId);
+        } else if (payload.sourceDeviceId) {
           await startOfferToClient(streamId, payload.sourceDeviceId);
           frameRelayStartTimer = setTimeout(() => {
             if (!webrtcConnected && !hasWebrtcEverConnected) {
@@ -780,7 +887,7 @@ const connectSocket = () => {
           await API.streams.end(streamId);
           await finalizeRecordingForStream(streamId, captureResult);
           stopFrameRelay();
-          if (socket && payload.sourceDeviceId) {
+          if (!isSfuMode() && socket && payload.sourceDeviceId) {
             socket.emit('webrtc:signal', {
               toDeviceId: payload.sourceDeviceId,
               streamSessionId: streamId,
@@ -816,19 +923,25 @@ const connectSocket = () => {
     });
     try {
       await API.streams.getSubscribeCreds(payload.streamSessionId);
+      if (isSfuMode()) {
+        await startSfuSubscribeHandshake(payload.streamSessionId);
+      }
       Toast.show('Connected to Stream', 'success');
+      if (!isSfuMode()) {
         remoteStreamWaitTimer = setTimeout(() => {
           if (!remoteClientStream) {
             Toast.show('Stream connected but no video received', 'error');
             addActivity('Stream', 'No remote video track received');
           }
         }, 6000);
+      }
     } catch (e) {
       Toast.show('Stream connect failed', 'error');
     }
   });
 
   socket.on('stream:frame', (payload) => {
+    if (isSfuMode()) return;
     if (webrtcConnected) return;
     if (!payload?.frame) return;
     if (remoteStreamWaitTimer) {
@@ -854,6 +967,7 @@ const connectSocket = () => {
   });
 
   socket.on('webrtc:signal', async (payload) => {
+    if (isSfuMode()) return;
     const device = store.get().device;
     if (!device || !payload?.streamSessionId || !payload?.signalType || !payload?.fromDeviceId) return;
 
@@ -986,6 +1100,7 @@ const Actions = {
       await API.auth.signIn({ email, password });
       const session = await API.auth.getSession();
       store.update({ session });
+      await detectMediaMode();
       Toast.show(`Welcome, ${session.user.name}`, 'success');
 
       // Proceed
@@ -1038,6 +1153,7 @@ const Actions = {
 
       store.update({ device: res.device, deviceToken: res.deviceToken });
       localStorage.setItem('mobileSimDevice', JSON.stringify({ device: res.device, deviceToken: res.deviceToken }));
+      await detectMediaMode();
 
       Toast.show('Device Registered', 'success');
       navigateBasedOnRole();