Matiss/Final-Year-Project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "feat(streams): add phase-2 SFU transport handshake and produce/consume APIs"

Browse Source 
This reverts commit 498a7f838b7747470b220701505c4bfbd3ea8cff.
This commit is contained in:
Matiss Jurevics
2026-02-20 14:00:00 +00:00
parent ef652ea7e5
commit 37d7c27ba0
12 changed files with 65 additions and 1395 deletions
Download Patch File Download Diff File Expand all files Collapse all files

349
Backend/public/mobile-sim.js
View File

@@ -138,21 +138,9 @@ 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: {
getRouterRtpCapabilities: (id) => API.request(`/streams/${id}/sfu/router-rtp-capabilities`),
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) }),
}
};
@@ -178,97 +166,10 @@ let webrtcConnected = false;
let hasWebrtcEverConnected = false;
let lastPeerConnectionState = null;
let pendingRemoteCandidates = [];
let mediaMode = 'legacy';
let mediasoupClientModulePromise = null;
let sfuDevice = null;
let sfuDeviceStreamSessionId = null;
let sfuSendTransport = null;
let sfuRecvTransport = null;
let sfuPublishedProducerId = null;
let sfuConsumedTrack = null;
let streamRequestInFlight = false;
let hasAutoRequestedInitialStream = false;
const inflightCameraStreamCommands = new Set();
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 loadMediasoupClientModule = async () => {
if (!mediasoupClientModulePromise) {
mediasoupClientModulePromise = import('https://cdn.jsdelivr.net/npm/mediasoup-client@3/+esm');
}
return await mediasoupClientModulePromise;
};
const closeSfuTransports = () => {
try {
sfuSendTransport?.close?.();
} catch { }
try {
sfuRecvTransport?.close?.();
} catch { }
if (sfuConsumedTrack) {
try {
sfuConsumedTrack.stop();
} catch { }
}
sfuSendTransport = null;
sfuRecvTransport = null;
sfuPublishedProducerId = null;
sfuConsumedTrack = null;
};
const resetSfuRuntime = () => {
closeSfuTransports();
sfuDevice = null;
sfuDeviceStreamSessionId = null;
};
const getMediasoupDeviceClass = async () => {
const module = await loadMediasoupClientModule();
return module.Device || module.default?.Device || module.default;
};
const ensureSfuDevice = async (streamSessionId) => {
if (sfuDevice && sfuDeviceStreamSessionId === streamSessionId) {
return sfuDevice;
}
resetSfuRuntime();
const DeviceClass = await getMediasoupDeviceClass();
if (!DeviceClass) {
throw new Error('mediasoup-client Device class unavailable');
}
const device = new DeviceClass();
const caps = await API.sfu.getRouterRtpCapabilities(streamSessionId);
if (!caps?.routerRtpCapabilities) {
throw new Error('Router RTP capabilities not available');
}
await device.load({ routerRtpCapabilities: caps.routerRtpCapabilities });
sfuDevice = device;
sfuDeviceStreamSessionId = streamSessionId;
return device;
};
const init = async () => {
// Load local storage
const saved = localStorage.getItem('mobileSimDevice');
@@ -283,7 +184,6 @@ 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();
@@ -613,7 +513,6 @@ const teardownPeerConnection = () => {
if (previousSessionId) {
pendingRemoteCandidates = pendingRemoteCandidates.filter((item) => item.streamSessionId !== previousSessionId);
}
closeSfuTransports();
clearClientStream();
};
@@ -761,146 +660,6 @@ 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 device = await ensureSfuDevice(streamSessionId);
const publishTransport = await API.sfu.createPublishTransport(streamSessionId);
const transportMeta = publishTransport?.transport;
const transportId = transportMeta?.transportId;
const transportOptions = transportMeta?.transportOptions;
if (!transportId || !transportOptions) {
throw new Error('Missing SFU publish transport parameters');
}
closeSfuTransports();
const sendTransport = device.createSendTransport(transportOptions);
sfuSendTransport = sendTransport;
sendTransport.on('connect', async ({ dtlsParameters }, callback, errback) => {
try {
await API.sfu.connectPublishTransport(streamSessionId, {
transportId,
dtlsParameters,
});
callback();
} catch (error) {
errback(error);
}
});
sendTransport.on('produce', async ({ kind, rtpParameters }, callback, errback) => {
try {
const produced = await API.sfu.produce(streamSessionId, {
transportId,
kind,
rtpParameters,
});
const producerId = produced?.producer?.producerId;
if (!producerId) {
throw new Error('SFU producer id missing');
}
sfuPublishedProducerId = producerId;
callback({ id: producerId });
} catch (error) {
errback(error);
}
});
const track = localCameraStream.getVideoTracks()[0];
if (!track) {
throw new Error('No local video track available for SFU publish');
}
await sendTransport.produce({ track });
addActivity('SFU', 'Publish handshake completed (mediasoup)');
};
const setClientSfuConnectedState = () => {
if (remoteStreamWaitTimer) {
clearTimeout(remoteStreamWaitTimer);
remoteStreamWaitTimer = null;
}
if (!remoteClientStream) {
setClientStreamPlaceholderText('Connected via SFU');
setClientStreamMode('connecting');
}
};
const startSfuSubscribeHandshake = async (streamSessionId) => {
const device = await ensureSfuDevice(streamSessionId);
const subscribeTransport = await API.sfu.createSubscribeTransport(streamSessionId);
const transportMeta = subscribeTransport?.transport;
const transportId = transportMeta?.transportId;
const transportOptions = transportMeta?.transportOptions;
if (!transportId || !transportOptions) {
throw new Error('Missing SFU subscribe transport parameters');
}
if (sfuRecvTransport) {
try {
sfuRecvTransport.close?.();
} catch { }
}
const recvTransport = device.createRecvTransport(transportOptions);
sfuRecvTransport = recvTransport;
recvTransport.on('connect', async ({ dtlsParameters }, callback, errback) => {
try {
await API.sfu.connectSubscribeTransport(streamSessionId, {
transportId,
dtlsParameters,
});
callback();
} catch (error) {
errback(error);
}
});
let consumeResult = null;
for (let attempt = 0; attempt < 8; attempt += 1) {
try {
consumeResult = await API.sfu.consume(streamSessionId, {
transportId,
rtpCapabilities: device.rtpCapabilities,
});
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');
}
const consumerMeta = consumeResult.consumer;
const consumer = await recvTransport.consume({
id: consumerMeta.consumerId,
producerId: consumerMeta.producerId,
kind: consumerMeta.kind,
rtpParameters: consumerMeta.rtpParameters,
});
sfuConsumedTrack = consumer.track;
const stream = new MediaStream([consumer.track]);
remoteClientStream = stream;
const videoEl = $('clientStreamVideo');
if (videoEl) {
videoEl.srcObject = stream;
setClientStreamMode('video');
void videoEl.play().catch(() => {});
}
setClientSfuConnectedState();
addActivity('SFU', `Subscribed to producer ${consumeResult.consumer.producerId} (mediasoup)`);
};
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const finalizeRecordingForStream = async (streamSessionId, captureResult) => {
@@ -917,14 +676,7 @@ const finalizeRecordingForStream = async (streamSessionId, captureResult) => {
if (recording?.id) {
try {
if (!captureResult?.blob || captureResult.blob.size === 0) {
const fallbackObjectKey = `sim/${streamSessionId}/${Date.now()}.webm`;
await API.events.finalizeRecording(recording.id, {
objectKey: fallbackObjectKey,
durationSeconds: captureResult?.durationSeconds ?? 15,
sizeBytes: captureResult?.blob?.size ?? 0,
});
addActivity('Recording', 'No local blob; finalized with simulator fallback');
return true;
throw new Error('No captured video blob to upload');
}
const uploadMeta = await API.request('/videos/upload-url', {
@@ -986,7 +738,6 @@ const connectSocket = () => {
socket.on('connect', () => {
store.update({ socketConnected: true });
addActivity('System', 'Connected to realtime server');
void detectMediaMode();
if (store.get().device?.role === 'camera') {
startCameraPreview();
}
@@ -997,7 +748,6 @@ const connectSocket = () => {
stopFrameRelay();
void stopLocalRecording();
teardownPeerConnection();
resetSfuRuntime();
store.update({ activeCameraDeviceId: null, activeStreamSessionId: null });
});
@@ -1008,31 +758,14 @@ const connectSocket = () => {
try {
if (payload.commandType === 'start_stream') {
const streamId = payload.payload.streamSessionId;
if (inflightCameraStreamCommands.has(streamId)) {
addActivity('Stream', `Duplicate start ignored for ${streamId.substring(0, 8)}`);
socket.emit('command:ack', { commandId: payload.commandId, status: 'acknowledged' });
return;
}
inflightCameraStreamCommands.add(streamId);
const ready = await startCameraPreview();
if (!ready) {
throw new Error('Camera permission is required before streaming');
}
try {
await API.streams.accept(streamId);
} catch (error) {
const message = error?.message || '';
if (!message.includes('status 409')) {
throw error;
}
addActivity('Stream', 'Accept already handled, continuing publish setup');
}
await API.streams.accept(streamId);
await API.streams.getPublishCreds(streamId);
await startLocalRecording();
if (isSfuMode()) {
await startSfuPublishHandshake(streamId);
} else if (payload.sourceDeviceId) {
if (payload.sourceDeviceId) {
await startOfferToClient(streamId, payload.sourceDeviceId);
frameRelayStartTimer = setTimeout(() => {
if (!webrtcConnected && !hasWebrtcEverConnected) {
@@ -1043,32 +776,25 @@ const connectSocket = () => {
addActivity('Stream', 'Accepted & Published');
// Auto-stop after 15s for simulation
setTimeout(async () => {
try {
const captureResult = await stopLocalRecording();
await API.streams.end(streamId);
await finalizeRecordingForStream(streamId, captureResult);
stopFrameRelay();
if (!isSfuMode() && socket && payload.sourceDeviceId) {
socket.emit('webrtc:signal', {
toDeviceId: payload.sourceDeviceId,
streamSessionId: streamId,
signalType: 'hangup',
});
}
teardownPeerConnection();
store.update({ activeCameraDeviceId: null, activeStreamSessionId: null });
addActivity('Stream', 'Ended auto-simulation');
} finally {
inflightCameraStreamCommands.delete(streamId);
const captureResult = await stopLocalRecording();
await API.streams.end(streamId);
await finalizeRecordingForStream(streamId, captureResult);
stopFrameRelay();
if (socket && payload.sourceDeviceId) {
socket.emit('webrtc:signal', {
toDeviceId: payload.sourceDeviceId,
streamSessionId: streamId,
signalType: 'hangup',
});
}
teardownPeerConnection();
store.update({ activeCameraDeviceId: null, activeStreamSessionId: null });
addActivity('Stream', 'Ended auto-simulation');
}, 15000);
}
socket.emit('command:ack', { commandId: payload.commandId, status: 'acknowledged' });
} catch (e) {
if (payload?.payload?.streamSessionId) {
inflightCameraStreamCommands.delete(payload.payload.streamSessionId);
}
socket.emit('command:ack', { commandId: payload.commandId, status: 'rejected', error: e.message });
}
});
@@ -1090,27 +816,19 @@ 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);
}
remoteStreamWaitTimer = setTimeout(() => {
if (!remoteClientStream) {
Toast.show('Stream connected but no video received', 'error');
addActivity('Stream', 'No remote video track received');
}
}, 6000);
} catch (e) {
const message = e?.message || 'Stream connect failed';
addActivity('SFU', message);
Toast.show('Stream connect failed', 'error');
}
});
socket.on('stream:frame', (payload) => {
if (isSfuMode()) return;
if (webrtcConnected) return;
if (!payload?.frame) return;
if (remoteStreamWaitTimer) {
@@ -1131,13 +849,11 @@ const connectSocket = () => {
socket.on('stream:ended', (payload) => {
if (payload?.streamSessionId && payload.streamSessionId === store.get().activeStreamSessionId) {
clearClientStream();
closeSfuTransports();
store.update({ activeCameraDeviceId: null, activeStreamSessionId: null });
}
});
socket.on('webrtc:signal', async (payload) => {
if (isSfuMode()) return;
const device = store.get().device;
if (!device || !payload?.streamSessionId || !payload?.signalType || !payload?.fromDeviceId) return;
@@ -1270,7 +986,6 @@ 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
@@ -1323,7 +1038,6 @@ 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();
@@ -1341,7 +1055,6 @@ const Actions = {
stopFrameRelay();
await stopLocalRecording();
teardownPeerConnection();
resetSfuRuntime();
stopCameraPreview();
localStorage.removeItem('mobileSimDevice');
Toast.show('Signed Out', 'info');
@@ -1380,25 +1093,12 @@ const Actions = {
},
requestStream: async (camId) => {
if (streamRequestInFlight) {
return;
}
const current = store.get();
if (current.activeStreamSessionId) {
return;
}
streamRequestInFlight = true;
try {
store.update({ activeCameraDeviceId: camId });
Toast.show('Requesting Stream...', 'info');
await API.streams.request(camId);
// Socket will handle the rest ('stream:started')
} catch (e) {
store.update({ activeCameraDeviceId: null });
} finally {
streamRequestInFlight = false;
}
} catch (e) { }
},
openRecording: async (recordingId) => {
@@ -1514,8 +1214,7 @@ const render = (state) => {
// 5. Client Mode Lists
if (state.device?.role === 'client' && state.screen === 'home') {
if (!hasAutoRequestedInitialStream && !state.activeCameraDeviceId && state.linkedCameras.length > 0 && !streamRequestInFlight) {
hasAutoRequestedInitialStream = true;
if (!state.activeCameraDeviceId && state.linkedCameras.length > 0) {
void Actions.requestStream(state.linkedCameras[0].cameraDeviceId);
}