diff --git a/src/api_productions.test.ts b/src/api_productions.test.ts index 40d7273..b7bd26a 100644 --- a/src/api_productions.test.ts +++ b/src/api_productions.test.ts @@ -168,6 +168,12 @@ const mockProductionManager = { .mockImplementation((lines: any[], id: string) => lines.find((l) => l.id === id) ), + requireLine: jest.fn().mockImplementation((lines: any[], id: string) => { + const found = lines.find((l) => l.id === id); + if (!found) throw new Error(`Line ${id} not found`); + return found; + }), + clearWhepSourceIfPinned: jest.fn().mockResolvedValue(undefined), updateUserLastSeen: jest .fn() .mockImplementation((sessionId: string) => sessionId === 'alive-session'), @@ -176,6 +182,7 @@ const mockProductionManager = { removeUserSession: jest .fn() .mockImplementation((sessionId: string) => sessionId), + emit: jest.fn(), createUserSession: jest.fn().mockResolvedValue(undefined), getActiveUsers: jest.fn().mockResolvedValue([]) } as any; @@ -483,6 +490,47 @@ describe('Production API', () => { }); }); + describe('PATCH /session/:id', () => { + test('flushes a 204 once the SDP answer is handled', async () => { + const patchSession = { + _id: 'mock-session', + name: 'usersession', + productionId: '1', + lineId: '1', + endpointId: 'mock-endpoint-1', + isActive: true, + isExpired: false, + isWhip: false, + sessionDescription: { audio: { ssrcs: [1] }, video: { ssrcs: [] } } + }; + const getSessionSpy = jest + .spyOn(mockDbManager, 'getSession') + .mockResolvedValue(patchSession as any); + mockProductionManager.getProduction = jest + .fn() + .mockResolvedValue(mockProductions[0]); + mockProductionManager.updateUserEndpoint = jest + .fn() + .mockResolvedValue(undefined); + mockProductionManager.updateSessionHasVideo = jest + .fn() + .mockResolvedValue(undefined); + mockCoreFunctions.handleAnswerRequest = jest + .fn() + .mockResolvedValue(undefined); + + const response = await server.inject({ + method: 'PATCH', + url: '/api/v1/session/mock-session', + body: { sdpAnswer: 'v=0' } + }); + + expect(response.statusCode).toBe(204); + expect(mockCoreFunctions.handleAnswerRequest).toHaveBeenCalled(); + getSessionSpy.mockRestore(); + }); + }); + describe('POST /production/:id/line/:id/participants', () => { test('can do long polling for change in line participants', async () => { mockProductionManager.once = jest diff --git a/src/api_productions.ts b/src/api_productions.ts index d4ac6d0..a76df8c 100644 --- a/src/api_productions.ts +++ b/src/api_productions.ts @@ -1,4 +1,4 @@ -import { Type } from '@sinclair/typebox'; +import { Static, Type } from '@sinclair/typebox'; import { FastifyPluginCallback } from 'fastify'; import { v4 as uuidv4 } from 'uuid'; import { CoreFunctions } from './api_productions_core_functions'; @@ -19,12 +19,16 @@ import { ProductionResponse, SdpAnswer, SessionResponse, + SetLineWhepSourceRequest, + SetLineWhepSourceResponse, + SetSessionVideoSourceRequest, + SetSessionVideoSourceResponse, SmbEndpointDescription, UserResponse, UserSession } from './models'; import { ProductionManager } from './production_manager'; -import { ISmbProtocol, SmbProtocol } from './smb'; +import { ISmbProtocol, SmbEndpointActionError, SmbProtocol } from './smb'; export interface ApiProductionsOptions { smbServerBaseUrl: string; @@ -38,10 +42,12 @@ export interface ApiProductionsOptions { function toUserResponse(doc: any) { const out: any = { - sessionId: (doc?._id ?? '').toString(), - name: (doc?.name ?? '').toString(), - isActive: !!doc?.isActive, - isWhip: !!doc?.isWhip + sessionId: String(doc?._id ?? ''), + name: String(doc?.name ?? ''), + isActive: Boolean(doc?.isActive), + isWhip: Boolean(doc?.isWhip), + isWhepReceiver: Boolean(doc?.isWhepReceiver), + hasVideo: Boolean(doc?.hasVideo) }; if (typeof doc?.endpointId === 'string' && doc.endpointId.length > 0) { out.endpointId = doc.endpointId; @@ -221,7 +227,8 @@ const apiProductions: FastifyPluginCallback = ( id: line.id, smbConferenceId: line.smbConferenceId, participants: sortParticipants(participants), - programOutputLine: line.programOutputLine || false + programOutputLine: line.programOutputLine || false, + videoEnabled: line.videoEnabled || false }; }); return { _id: production._id, name: production.name, lines }; @@ -413,7 +420,8 @@ const apiProductions: FastifyPluginCallback = ( id: line.id, smbConferenceId: line.smbConferenceId, participants: sortParticipants(participants), - programOutputLine: line.programOutputLine || false + programOutputLine: line.programOutputLine || false, + videoEnabled: line.videoEnabled || false }; } ); @@ -457,7 +465,8 @@ const apiProductions: FastifyPluginCallback = ( await productionManager.addProductionLine( production, request.body.name, - request.body.programOutputLine || false + request.body.programOutputLine || false, + request.body.videoEnabled || false ); const allLinesResponse: LineResponse[] = await coreFunctions.getAllLinesResponse(production); @@ -509,8 +518,10 @@ const apiProductions: FastifyPluginCallback = ( sessionId: (s._id ?? '').toString(), endpointId: s.endpointId, name: s.name, - isActive: !!s.isActive, - isWhip: s.isWhip + isActive: s.isWhip ? true : Boolean(s.isActive), + isWhip: Boolean(s.isWhip), + isWhepReceiver: Boolean(s.isWhepReceiver), + hasVideo: Boolean(s.hasVideo) })); const lineResponse: LineResponse = { @@ -518,7 +529,9 @@ const apiProductions: FastifyPluginCallback = ( id: line.id, smbConferenceId: line.smbConferenceId, participants: sortParticipants(participants), - programOutputLine: line.programOutputLine || false + programOutputLine: line.programOutputLine || false, + videoEnabled: line.videoEnabled || false, + whepSourceSessionId: line.whepSourceSessionId ?? null }; reply.code(200).send(lineResponse); } catch (err) { @@ -585,7 +598,8 @@ const apiProductions: FastifyPluginCallback = ( reply.code(200).send({ name: request.body.name, id: lineId, - programOutputLine: line.programOutputLine || false + programOutputLine: line.programOutputLine || false, + videoEnabled: line.videoEnabled || false }); } } @@ -597,6 +611,215 @@ const apiProductions: FastifyPluginCallback = ( } ); + fastify.patch<{ + Params: { productionId: string; lineId: string }; + Body: Static; + Reply: Static | ErrorResponse | string; + }>( + '/production/:productionId/line/:lineId/whep-source', + { + schema: { + description: + 'Pin a single participant as the WHEP egress source for this line. Pass `null` to clear and restore forward-all behaviour.', + body: SetLineWhepSourceRequest, + response: { + 200: SetLineWhepSourceResponse, + 404: ErrorResponse, + 500: Type.String() + } + } + }, + async (request, reply) => { + try { + const { productionId, lineId } = request.params; + const { pinnedSessionId: rawPinned } = request.body; + const pinnedSessionId: string | null = + rawPinned === '' ? null : rawPinned; + + let production; + try { + production = await productionManager.requireProduction( + parseInt(productionId, 10) + ); + } catch { + reply + .code(404) + .send({ message: `Production with id ${productionId} not found` }); + return; + } + + const line = productionManager.getLine(production.lines, lineId); + if (!line) { + reply.code(404).send({ message: `Line with id ${lineId} not found` }); + return; + } + + const updated = await productionManager.setLineWhepSource( + production, + lineId, + pinnedSessionId + ); + if (!updated) { + reply.code(500).send('Failed to update WHEP source pin'); + return; + } + + reply.code(200).send({ lineId, pinnedSessionId }); + } catch (err) { + Log().error(err); + reply + .code(500) + .send('Exception thrown when trying to set WHEP source: ' + err); + } + } + ); + + fastify.patch<{ + Params: { sessionId: string }; + Body: Static; + Reply: + | Static + | ErrorResponse + | string; + }>( + '/session/:sessionId/video-source', + { + schema: { + description: + 'Pin a single publisher as this session’s video source. SMB egress filter updates in place via the `reconfigure` action. Pass `null` to clear.', + body: SetSessionVideoSourceRequest, + response: { + 200: SetSessionVideoSourceResponse, + 404: ErrorResponse, + 409: ErrorResponse, + 425: ErrorResponse, + 500: Type.String() + } + } + }, + async (request, reply) => { + try { + const { sessionId } = request.params; + const { pinnedSessionId: rawPinned } = request.body; + const pinnedSessionId: string | null = + rawPinned === '' ? null : rawPinned; + + const userSession = await dbManager.getSession(sessionId); + if (!userSession) { + reply.code(404).send({ message: `Session ${sessionId} not found` }); + return; + } + + const endpointId = userSession.endpointId; + const endpointDescription = userSession.sessionDescription; + if (!endpointId || !endpointDescription) { + reply.code(409).send({ + message: + 'Session has no SMB endpoint yet (still negotiating). Try again after PATCH /session.' + }); + return; + } + + let whitelist: number[] = []; + if (pinnedSessionId) { + const sourceSession = await dbManager.getSession(pinnedSessionId); + const sourceVideo: any = sourceSession?.sessionDescription?.video; + const ssrcs: number[] = Array.isArray(sourceVideo?.ssrcs) + ? sourceVideo.ssrcs + : []; + whitelist = Array.from(new Set(ssrcs)) + .filter((n) => Number.isFinite(n)) + .slice(0, 2); + if (whitelist.length === 0) { + reply.code(425).send({ + message: + `Pin source ${pinnedSessionId} has no video SSRCs yet ` + + `(still negotiating). Retry shortly.` + }); + return; + } + } + + const updatedDescription: SmbEndpointDescription = JSON.parse( + JSON.stringify(endpointDescription) + ); + if (updatedDescription.video) { + if (whitelist.length > 0) { + updatedDescription.video['ssrc-whitelist'] = whitelist; + } else { + delete updatedDescription.video['ssrc-whitelist']; + } + } + + const productionIdNum = parseInt(userSession.productionId, 10); + let production; + try { + production = await productionManager.requireProduction( + productionIdNum + ); + } catch { + reply.code(404).send({ message: 'Production not found' }); + return; + } + const line = productionManager.requireLine( + production.lines, + userSession.lineId + ); + + try { + await smb.reconfigureEndpoint( + smbServerUrl, + line.smbConferenceId, + endpointId, + updatedDescription, + smbServerApiKey + ); + } catch (err) { + if ( + err instanceof SmbEndpointActionError && + err.isEndpointNotConfiguredYet + ) { + reply.code(425).send({ + message: + `Session ${sessionId} has no configured SMB endpoint yet ` + + `(still negotiating). Retry shortly.` + }); + return; + } + throw err; + } + + const pinChanged = + pinnedSessionId !== null && + pinnedSessionId !== (userSession.pinnedVideoSessionId ?? null); + if (pinChanged) { + await smb.requestKeyframe( + smbServerUrl, + line.smbConferenceId, + endpointId, + updatedDescription, + smbServerApiKey + ); + } + + await productionManager.updateSessionVideoPin( + sessionId, + updatedDescription, + pinnedSessionId + ); + + reply.code(200).send({ sessionId, pinnedSessionId }); + } catch (err) { + Log().error(err); + reply + .code(500) + .send( + 'Exception thrown when trying to set session video source: ' + err + ); + } + } + ); + fastify.delete<{ Params: { productionId: string; lineId: string }; Reply: string | ErrorResponse; @@ -677,6 +900,13 @@ const apiProductions: FastifyPluginCallback = ( lineId ); + // Look up videoEnabled from the line configuration + const production = await productionManager.requireProduction( + parseInt(productionId, 10) + ); + const line = productionManager.requireLine(production.lines, lineId); + const videoEnabled = line.videoEnabled ?? false; + await productionManager.createUserSession( smbConferenceId, productionId, @@ -695,10 +925,12 @@ const apiProductions: FastifyPluginCallback = ( smbConferenceId, endpointId, true, // audio + videoEnabled, // video true, // data true, // iceControlling - 'ssrc-rewrite', // relayType - isNaN(idleTimeout) ? 60 : idleTimeout + 'ssrc-rewrite', // audio relay type + isNaN(idleTimeout) ? 60 : idleTimeout, + 'ssrc-rewrite' ); if (!endpoint.audio) { throw new Error('Missing audio when creating sdp offer for endpoint'); @@ -721,7 +953,8 @@ const apiProductions: FastifyPluginCallback = ( endpoint, username, endpointId, - sessionId + sessionId, + videoEnabled ); if (sdpOffer) { @@ -754,7 +987,7 @@ const apiProductions: FastifyPluginCallback = ( 'Provide client local SDP description as request body to finalize connection protocol.', params: SessionIdParams, response: { - 200: Type.String(), + 204: Type.Null(), 400: Type.String(), 500: Type.String() } @@ -813,6 +1046,53 @@ const apiProductions: FastifyPluginCallback = ( throw new Error('Could not get connection endpoint id'); } + let subscribeToVideo: + | { ssrcs: number[]; endpointId: string } + | undefined; + try { + const productionIdNum = parseInt(userSession.productionId, 10); + if (!Number.isNaN(productionIdNum)) { + const production = await productionManager.getProduction( + productionIdNum + ); + const lineForPin = production?.lines.find( + (l) => l.id === userSession.lineId + ); + let pinnedSessionId: string | null = + lineForPin?.whepSourceSessionId ?? null; + + if (!pinnedSessionId) { + const whipCandidates = (await dbManager.getSessionsByQuery({ + productionId: userSession.productionId, + lineId: userSession.lineId, + isActive: true, + isWhip: true, + hasVideo: true + } as Partial)) as UserSession[]; + if (whipCandidates.length === 1) { + pinnedSessionId = + ( + whipCandidates[0] as UserSession & { _id?: unknown } + )._id?.toString?.() ?? null; + } + } + + if (pinnedSessionId) { + const sourceSession = await dbManager.getSession(pinnedSessionId); + const sourceVideo: any = sourceSession?.sessionDescription?.video; + const sourceEndpointId = sourceSession?.endpointId; + const ssrcs: number[] = Array.isArray(sourceVideo?.ssrcs) + ? sourceVideo.ssrcs + : []; + if (sourceEndpointId && ssrcs.length > 0) { + subscribeToVideo = { ssrcs, endpointId: sourceEndpointId }; + } + } + } + } catch { + // Pin resolution failed — fall back to default rotation. + } + await coreFunctions.handleAnswerRequest( smb, smbServerUrl, @@ -820,8 +1100,29 @@ const apiProductions: FastifyPluginCallback = ( line.smbConferenceId, endpointId, connectionEndpointDescription, - request.body.sdpAnswer + request.body.sdpAnswer, + subscribeToVideo ); + + await productionManager.updateUserEndpoint( + sessionId, + endpointId, + connectionEndpointDescription + ); + + try { + const sendingSsrcs = connectionEndpointDescription.video?.ssrcs ?? []; + await productionManager.updateSessionHasVideo( + sessionId, + sendingSsrcs.length > 0 + ); + } catch (hasVideoErr) { + Log().warn( + `Could not determine hasVideo for session=${sessionId} from ` + + `endpoint: ${hasVideoErr}` + ); + } + reply.code(204).send(); } catch (err) { Log().error(err); @@ -883,10 +1184,59 @@ const apiProductions: FastifyPluginCallback = ( async (request, reply) => { const sessionId = request.params.sessionId; try { - const deletedSessionId = await dbManager.deleteUserSession(sessionId); - if (!deletedSessionId) { + await productionManager.clearWhepSourceIfPinned(sessionId); + + try { + const affected = await productionManager.getReceiversPinnedToSession( + sessionId + ); + if (affected.length > 0) { + const production = await productionManager.getProduction( + parseInt(affected[0].productionId, 10) + ); + const line = production?.lines.find( + (l) => l.id === affected[0].lineId + ); + if (line) { + await Promise.all( + affected.map(async (receiver) => { + const receiverId = (receiver as any)._id?.toString?.(); + const endpointId = receiver.endpointId; + const endpointDescription = receiver.sessionDescription; + if (!receiverId || !endpointId || !endpointDescription) + return; + const updatedDescription: SmbEndpointDescription = JSON.parse( + JSON.stringify(endpointDescription) + ); + if (updatedDescription.video) { + delete updatedDescription.video['ssrc-whitelist']; + } + await smb.reconfigureEndpoint( + smbServerUrl, + line.smbConferenceId, + endpointId, + updatedDescription, + smbServerApiKey + ); + await productionManager.updateSessionVideoPin( + receiverId, + updatedDescription, + null + ); + }) + ); + } + } + } catch { + // Never let pin reconciliation block the session delete itself. + } + + const ok = await dbManager.deleteUserSession(sessionId); + if (!ok) { throw new Error(`Could not delete connection ${sessionId}`); } + productionManager.removeUserSession(sessionId); + productionManager.emit('users:change'); reply.code(200).send(`Deleted connection ${sessionId}`); } catch (err) { Log().error(err); @@ -943,8 +1293,10 @@ const apiProductions: FastifyPluginCallback = ( sessionId: s._id.toString(), endpointId: s.endpointId, name: s.name, - isActive: !!s.isActive, - isWhip: !!s.isWhip + isActive: s.isWhip ? true : Boolean(s.isActive), + isWhip: Boolean(s.isWhip), + isWhepReceiver: Boolean(s.isWhepReceiver), + hasVideo: Boolean(s.hasVideo) })); reply.code(200).send(sortParticipants(participants)); @@ -982,6 +1334,20 @@ const apiProductions: FastifyPluginCallback = ( } ); + fastify.get<{ Params: { sessionId: string } }>( + '/session/:sessionId/name', + async (request, reply) => { + const name = await productionManager.getUserNameBySessionId( + request.params.sessionId + ); + if (name == null) { + reply.code(404).send({ message: 'Session not found' }); + return; + } + reply.code(200).send({ sessionId: request.params.sessionId, name }); + } + ); + next(); }; diff --git a/src/api_productions_codec_negotiation.test.ts b/src/api_productions_codec_negotiation.test.ts new file mode 100644 index 0000000..5259a9f --- /dev/null +++ b/src/api_productions_codec_negotiation.test.ts @@ -0,0 +1,188 @@ +jest.mock('./log', () => ({ + Log: () => ({ + error: jest.fn(), + warn: jest.fn(), + info: jest.fn(), + debug: jest.fn() + }) +})); + +import { parse } from 'sdp-transform'; +import { + CoreFunctions, + selectVideoCodec, + smbAdvertisedVideoCodecs +} from './api_productions_core_functions'; +import { ConnectionQueue } from './connection_queue'; +import { RtpCodec } from './media_streams_info'; +import { MockSmbProtocol } from './mock-smb-protocol'; +import { SmbEndpointDescription } from './models'; +import { ProductionManager } from './production_manager'; +import { + audioVideoOffer, + createMockEndpointDescription +} from './test-fixtures/sdp-fixtures'; + +// A codec must be one SMB advertised, not merely one the client offered: +// SMB's default is VP8 while every browser and whip-mpegts offers H264, so +// negotiating from the offer alone leaves receivers with audio and no video. + +const smbUrl = 'http://smb.test'; +const smbKey = 'key'; + +/** An SMB allocation whose video block advertises exactly `codecs` (+ RTX). */ +const endpointAdvertising = (codecs: string[]): SmbEndpointDescription => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const endpoint = createMockEndpointDescription() as any; + endpoint.video['payload-types'] = [ + ...codecs.map((name, index) => ({ + id: 100 + index, + name, + clockrate: 90000, + parameters: {}, + 'rtcp-fbs': [{ type: 'nack', subtype: '' }] + })), + { + id: 120, + name: 'rtx', + clockrate: 90000, + parameters: { apt: '100' }, + 'rtcp-fbs': [] + } + ]; + return endpoint as SmbEndpointDescription; +}; + +const rtp = (codec: string, payload: number): RtpCodec => + ({ codec, payload, rate: 90000 } as RtpCodec); + +describe('video codec negotiation', () => { + describe('smbAdvertisedVideoCodecs', () => { + test('reports the advertised codecs, uppercased, without RTX', () => { + expect(smbAdvertisedVideoCodecs(endpointAdvertising(['vp8']))).toEqual([ + 'VP8' + ]); + }); + + test('is empty when the allocation carried no video payload-types', () => { + expect(smbAdvertisedVideoCodecs(createMockEndpointDescription())).toEqual( + [] + ); + }); + }); + + describe('selectVideoCodec', () => { + const offered = [rtp('VP8', 96), rtp('H264', 98)]; + + test('picks VP8 when the bridge only advertises VP8', () => { + expect(selectVideoCodec(offered, ['VP8'])?.codec).toBe('VP8'); + }); + + test('picks H264 when the bridge advertises H264', () => { + expect(selectVideoCodec(offered, ['H264'])?.codec).toBe('H264'); + }); + + test('prefers H264 when the bridge advertises both', () => { + expect(selectVideoCodec(offered, ['VP8', 'H264'])?.codec).toBe('H264'); + }); + + test('falls back to pipeline preference when SMB advertised nothing', () => { + expect(selectVideoCodec(offered, [])?.codec).toBe('H264'); + }); + + test('returns undefined when offer and bridge share no codec', () => { + expect(selectVideoCodec([rtp('H264', 98)], ['VP8'])).toBeUndefined(); + }); + }); + + describe('against a VP8-only bridge (regression)', () => { + let coreFunctions: CoreFunctions; + let mockSmb: MockSmbProtocol; + + beforeEach(() => { + mockSmb = new MockSmbProtocol(); + coreFunctions = new CoreFunctions( + {} as ProductionManager, + new ConnectionQueue() + ); + }); + + test('configures SMB with VP8 even though the offer also carries H264', async () => { + const confId = await mockSmb.allocateConference(smbUrl, smbKey); + const endpoint = endpointAdvertising(['VP8']); + + await coreFunctions.configureEndpointForWhipWhep( + audioVideoOffer(), + endpoint, + mockSmb, + smbUrl, + smbKey, + confId, + 'ep-vp8' + ); + + const configured = mockSmb.getEndpoint(confId, 'ep-vp8'); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((configured as any)?.video['payload-type'].name).toBe('VP8'); + }); + + test('answers the publisher with VP8 only, so it cannot encode H264', async () => { + const endpoint = endpointAdvertising(['VP8']); + + const answer = await coreFunctions.createWhipWhepAnswer( + audioVideoOffer(), + endpoint + ); + + const video = parse(answer).media.find((m) => m.type === 'video'); + const codecs = (video?.rtp ?? []).map((r) => r.codec.toUpperCase()); + expect(codecs).toContain('VP8'); + expect(codecs).not.toContain('H264'); + }); + + test('rejects an H264-only publisher instead of negotiating a dead stream', async () => { + const confId = await mockSmb.allocateConference(smbUrl, smbKey); + const endpoint = endpointAdvertising(['VP8']); + const offer = audioVideoOffer(); + const video = offer.media.find((m) => m.type === 'video'); + // Drop VP8 from the offer: an H264-only publisher, e.g. whip-mpegts. + if (video) { + video.rtp = video.rtp.filter( + (r) => r.codec.toUpperCase() !== 'VP8' + ) as typeof video.rtp; + } + + await expect( + coreFunctions.configureEndpointForWhipWhep( + offer, + endpoint, + mockSmb, + smbUrl, + smbKey, + confId, + 'ep-h264-only' + ) + ).rejects.toThrow(/No video codec in common.*SMB advertises: VP8/s); + }); + }); + + describe('against an H264 bridge', () => { + test('still negotiates H264, so existing deployments are unchanged', async () => { + const endpoint = endpointAdvertising(['H264']); + const coreFunctions = new CoreFunctions( + {} as ProductionManager, + new ConnectionQueue() + ); + + const answer = await coreFunctions.createWhipWhepAnswer( + audioVideoOffer(), + endpoint + ); + + const video = parse(answer).media.find((m) => m.type === 'video'); + const codecs = (video?.rtp ?? []).map((r) => r.codec.toUpperCase()); + expect(codecs).toContain('H264'); + expect(codecs).not.toContain('VP8'); + }); + }); +}); diff --git a/src/api_productions_core_functions.test.ts b/src/api_productions_core_functions.test.ts index a60e9cc..eb18aac 100644 --- a/src/api_productions_core_functions.test.ts +++ b/src/api_productions_core_functions.test.ts @@ -260,11 +260,11 @@ describe('CoreFunctions SDP Tests', () => { const configured = mockSmb.getEndpoint(confId, 'ep-8'); expect(configured?.video).toBeDefined(); - expect(configured?.video?.['payload-type'].name).toBe('VP8'); - expect(configured?.video?.['payload-type'].clockrate).toBe(90000); + expect(configured?.video?.['payload-type']?.name).toBe('H264'); + expect(configured?.video?.['payload-type']?.clockrate).toBe(90000); }); - test('filters video codecs to supported only (VP8/H264/VP9)', async () => { + test('filters video codecs to supported only (VP8/H264)', async () => { const confId = await mockSmb.allocateConference(smbUrl, smbKey); const endpoint = createMockEndpointDescription(); const offer = audioVideoOffer(); @@ -280,8 +280,9 @@ describe('CoreFunctions SDP Tests', () => { ); const configured = mockSmb.getEndpoint(confId, 'ep-9'); - // rtx (97) is not VP8/H264/VP9, but VP8 (96) and H264 (98) are - expect(configured?.video?.['payload-type'].id).toBe(96); + // H264 (98) is preferred over VP8 (96); rtx (97/99) and VP9 are + // filtered out of the supported set. + expect(configured?.video?.['payload-type']?.id).toBe(98); }); test('filters video rtp-hdrexts to abs-send-time and rtp-stream-id', async () => { @@ -431,6 +432,51 @@ describe('CoreFunctions SDP Tests', () => { // Original should be unchanged expect(endpoint['bundle-transport']!.ice!.ufrag).toBe(originalUfrag); }); + + test('blocks video egress to a WHIP publisher with an empty ssrc-whitelist', async () => { + const confId = await mockSmb.allocateConference(smbUrl, smbKey); + const endpoint = createMockEndpointDescription(); + const offer = audioVideoOffer(); + + // receiveOnly omitted — this is the WHIP publisher path. + await coreFunctions.configureEndpointForWhipWhep( + offer, + endpoint, + mockSmb, + smbUrl, + smbKey, + confId, + 'ep-whip-egress' + ); + + const configured = mockSmb.getEndpoint(confId, 'ep-whip-egress'); + // Present, and empty — not absent, which would mean forward-everything. + expect(configured?.video?.['ssrc-whitelist']).toEqual([]); + // Ingress must be untouched: SMB still needs this publisher's own video. + expect(configured?.video?.streams?.length).toBeGreaterThan(0); + expect(configured?.video?.ssrcs?.length).toBeGreaterThan(0); + }); + + test('does not block video egress for a receive-only WHEP endpoint', async () => { + const confId = await mockSmb.allocateConference(smbUrl, smbKey); + const endpoint = createMockEndpointDescription(); + const offer = audioVideoOffer(); + + await coreFunctions.configureEndpointForWhipWhep( + offer, + endpoint, + mockSmb, + smbUrl, + smbKey, + confId, + 'ep-whep-egress', + true // receiveOnly + ); + + const configured = mockSmb.getEndpoint(confId, 'ep-whep-egress'); + // Unpinned WHEP must fall back to last-N, so the key must stay absent. + expect(configured?.video?.['ssrc-whitelist']).toBeUndefined(); + }); }); // ══════════════════════════════════════════════════════════════════ @@ -552,7 +598,7 @@ describe('CoreFunctions SDP Tests', () => { expect(audioMedia?.direction).toBe('recvonly'); }); - test('filters video to VP8 + RTX only', async () => { + test('filters video to H264 + RTX only', async () => { const offer = audioVideoOffer(); const endpoint = createMockEndpointDescription(); @@ -563,12 +609,12 @@ describe('CoreFunctions SDP Tests', () => { const parsed = parse(sdpAnswer); const videoMedia = parsed.media.find((m) => m.type === 'video'); - // VP8 (96) and RTX (97), not H264 (98) + // H264 (98) is preferred over VP8 (96); its RTX (99, apt=98) is kept. expect(videoMedia?.rtp.length).toBe(2); const codecs = videoMedia?.rtp.map((r) => r.codec); - expect(codecs).toContain('VP8'); + expect(codecs).toContain('H264'); expect(codecs).toContain('rtx'); - expect(codecs).not.toContain('H264'); + expect(codecs).not.toContain('VP8'); }); test('sets BUNDLE group with all media mids', async () => { @@ -873,6 +919,7 @@ describe('CoreFunctions SDP Tests', () => { 'ep-1', true, false, + false, true, 'ssrc-rewrite', 60 @@ -893,6 +940,7 @@ describe('CoreFunctions SDP Tests', () => { 'ep-1', true, false, + false, true, 'ssrc-rewrite', 60 diff --git a/src/api_productions_core_functions.ts b/src/api_productions_core_functions.ts index be29106..4d719e9 100644 --- a/src/api_productions_core_functions.ts +++ b/src/api_productions_core_functions.ts @@ -17,8 +17,50 @@ import { import { Log } from './log'; import { LineResponse, Production, SmbEndpointDescription } from './models'; import { ProductionManager } from './production_manager'; +import { + NORMALIZED_VIDEO_PT_MAIN, + NORMALIZED_VIDEO_PT_RTX +} from './sfu/constants'; import { ISmbProtocol } from './smb'; +/** Video codecs supported through the pipeline, in preference order. */ +const SUPPORTED_VIDEO_CODECS = ['H264', 'VP8']; + +/** + * The video codec names SMB advertised for this endpoint, RTX excluded. + * Empty when the allocation carried no video payload-types. + */ +export function smbAdvertisedVideoCodecs( + endpoint: SmbEndpointDescription +): string[] { + return (endpoint.video?.['payload-types'] ?? []) + .map((pt) => pt.name?.toUpperCase()) + .filter((name): name is string => !!name && name !== 'RTX'); +} + +/** + * The most preferred video codec present in both the offer and SMB's + * advertised codecs. Falls back to the pipeline preference order when SMB + * advertised none. Undefined when the two share no supported codec. + */ +export function selectVideoCodec( + offered: RtpCodec[], + smbCodecs: string[] +): RtpCodec | undefined { + const allowed = + smbCodecs.length > 0 + ? SUPPORTED_VIDEO_CODECS.filter((codec) => smbCodecs.includes(codec)) + : SUPPORTED_VIDEO_CODECS; + + for (const name of allowed) { + const match = offered.find( + (rtp: RtpCodec) => rtp.codec.toUpperCase() === name + ); + if (match) return match; + } + return undefined; +} + export class CoreFunctions { private productionManager: ProductionManager; private connectionQueue: ConnectionQueue; @@ -38,7 +80,8 @@ export class CoreFunctions { endpoint: SmbEndpointDescription, username: string, endpointId: string, - sessionId: string + sessionId: string, + videoEnabled = false ): Promise { if (!endpoint.audio) { throw new Error('Missing audio when creating offer'); @@ -54,10 +97,21 @@ export class CoreFunctions { }); }); + const videoSsrcs: MediaStreamsInfoSsrc[] = + videoEnabled && endpoint.video?.ssrcs?.length + ? endpoint.video.ssrcs.map((ssrcNr) => ({ + ssrc: ssrcNr.toString(), + cname: uuidv4(), + mslabel: uuidv4(), + label: uuidv4() + })) + : []; + const endpointMediaStreamInfo = { audio: { ssrcs: ssrcs - } + }, + ...(videoEnabled && endpoint.video && { video: { ssrcs: videoSsrcs } }) }; const connection = new Connection( @@ -95,23 +149,57 @@ export class CoreFunctions { lineId: string, endpointId: string, audio: boolean, + video: boolean, data: boolean, iceControlling: boolean, relayType: 'ssrc-rewrite' | 'forwarder' | 'mixed', - endpointIdleTimeout: number + endpointIdleTimeout: number, + videoRelayType?: 'ssrc-rewrite' | 'forwarder' | 'mixed' ): Promise { const endpoint: SmbEndpointDescription = await smb.allocateEndpoint( smbServerUrl, lineId, endpointId, audio, + video, data, iceControlling, relayType, endpointIdleTimeout, - smbServerApiKey + smbServerApiKey, + videoRelayType ); + // Normalize video payload type numbers to stable values so the SDP offer, + // browser answer, and SMB configure body all agree. Both H264 and VP8 + // normalize to PT 96/97 — they are mutually exclusive. Using PT 96 matches + // whip-mpegts's native H264 PT so forwarder-mode receivers see a consistent PT. + // Also corrects SMB's RTX apt which may point to the wrong PT. + const pts = endpoint.video?.['payload-types']; + if (pts) { + const h264 = pts.find((pt) => pt.name.toUpperCase() === 'H264'); + const vp8 = pts.find((pt) => pt.name.toUpperCase() === 'VP8'); + const preferred = h264 ?? vp8; + if (preferred) { + const mainPt = NORMALIZED_VIDEO_PT_MAIN; + const rtxPt = NORMALIZED_VIDEO_PT_RTX; + preferred.id = mainPt; + const rtx = pts.find((pt) => pt.name.toLowerCase() === 'rtx'); + if (rtx) { + rtx.id = rtxPt; + if (rtx.parameters?.['apt'] !== undefined) + rtx.parameters['apt'] = String(mainPt); + } + } + // Force H264 profile-level-id to Constrained Baseline (42e01f). SMB + // reports pure Baseline (42001f) which Safari's WebRTC stack rejects, + // collapsing the video m-line to port 0 in the answer. CBP is the only + // H264 profile WebRTC mandates (RFC 7742) and is universally decodable. + if (h264?.parameters) { + h264.parameters['profile-level-id'] = '42e01f'; + } + } + return endpoint; } @@ -122,7 +210,20 @@ export class CoreFunctions { smbServerUrl: string, smbServerApiKey: string, smbConferenceId: string, - endpointId: string + endpointId: string, + receiveOnly = false, + /** + * When set on a receive-only (WHEP) endpoint, declare to SMB that this + * endpoint subscribes to the named publisher's video stream — limiting + * forwarding to that one source instead of the SFU-default forward-all. + * + * Ignored when `receiveOnly` is false. + */ + subscribeToVideo?: { + streams: any[]; + ssrcs: number[]; + endpointId: string; + } ): Promise { const offer: SessionDescription = JSON.parse(JSON.stringify(sdpOffer)); const endpoint: SmbEndpointDescription = JSON.parse( @@ -141,25 +242,42 @@ export class CoreFunctions { throw new Error('Missing ice in endpointDescription'); } - const audioMedia = { - ...offer.media.find((media) => media.type === 'audio') - } as MediaDescription; - - transport.ice.ufrag = offer.iceUfrag ?? audioMedia?.iceUfrag ?? ''; - transport.ice.pwd = offer.icePwd ?? audioMedia?.icePwd ?? ''; + // The bundle-transport (ICE/DTLS) is carried on whichever m-line has + // fingerprint/iceUfrag — usually audio, but a port-0 audio reject or + // a data-first ordering can shift it. Original code used + // `{...find(...)}` which spreads to `{}` (always truthy) so the + // short-circuit incorrectly fell through to offer.media[0] (data + // m-line) and emptied transport. Find by attribute presence instead. + const transportMedia = + (offer.media.find((m) => m.fingerprint || m.iceUfrag) as + | MediaDescription + | undefined) ?? (offer.media[0] as MediaDescription | undefined); + + transport.ice.ufrag = offer.iceUfrag ?? transportMedia?.iceUfrag ?? ''; + transport.ice.pwd = offer.icePwd ?? transportMedia?.icePwd ?? ''; transport.dtls.hash = - offer.fingerprint?.hash ?? audioMedia?.fingerprint?.hash ?? ''; + offer.fingerprint?.hash ?? transportMedia?.fingerprint?.hash ?? ''; transport.dtls.type = - offer.fingerprint?.type ?? audioMedia?.fingerprint?.type ?? ''; - transport.dtls.setup = offer.setup ?? audioMedia?.setup ?? ''; + offer.fingerprint?.type ?? transportMedia?.fingerprint?.type ?? ''; + transport.dtls.setup = offer.setup ?? transportMedia?.setup ?? ''; + + if (!transport.dtls.hash || !transport.dtls.type) { + throw new Error( + `Missing DTLS fingerprint in offer (would result in null cipher). ` + + `offer.fingerprint=${JSON.stringify(offer.fingerprint)}, ` + + `mediaFingerprints=${JSON.stringify( + offer.media.map((m) => m.fingerprint) + )}` + ); + } if (!transport.ice.candidates || transport.ice.candidates.length === 0) { throw new Error('ICE candidates missing in transport'); } - transport.ice.candidates = !audioMedia.candidates + transport.ice.candidates = !transportMedia?.candidates ? [] - : audioMedia.candidates.flatMap((element) => { + : transportMedia.candidates.flatMap((element) => { return { generation: element.generation ? element.generation : 0, component: element.component, @@ -184,6 +302,12 @@ export class CoreFunctions { media.ssrcs ?.filter((ssrc) => ssrc.attribute === 'msid') .forEach((ssrc) => endpoint.audio.ssrcs.push(parseInt(`${ssrc.id}`))); + if (!media.rtp?.[0]) { + throw new Error( + 'Audio m-line in offer has no rtp payload entries — rejected ' + + 'or malformed audio m-line cannot be configured.' + ); + } endpoint.audio['payload-type'].id = media.rtp[0].payload; endpoint.audio['rtp-hdrexts'] = []; media.ext?.forEach((ext: RtpHeaderExt) => @@ -201,7 +325,7 @@ export class CoreFunctions { if (!smbVideoStream) { smbVideoStream = { sources: [], - id: mediaStreamId, + id: receiveOnly ? mediaStreamId : endpointId, content: 'video' }; streamsMap.set(mediaStreamId, smbVideoStream); @@ -215,11 +339,13 @@ export class CoreFunctions { if (feedbackGroup) { const ssrcsSplit = feedbackGroup.ssrcs.split(' '); if (`${ssrc.id}` === ssrcsSplit[0]) { + const main = parseInt(ssrcsSplit[0]); + // Skip feedback when the FID group has only one SSRC — + // otherwise parseInt(undefined) ships feedback: NaN to SMB. smbVideoStream.sources = [ - { - main: parseInt(ssrcsSplit[0]), - feedback: parseInt(ssrcsSplit[1]) - } + ssrcsSplit.length >= 2 + ? { main, feedback: parseInt(ssrcsSplit[1]) } + : { main } ]; } } else { @@ -231,8 +357,57 @@ export class CoreFunctions { } }); - streamsMap.forEach((value) => videoStreams.push(value)); - const supportedCodecs = ['VP8', 'H264', 'VP9']; + // Only collect sender SSRCs into videoStreams for WHIP endpoints. + // WHEP offers may include a=ssrc: lines (Chrome UA hints for receive + // tracks) that must not be treated as sender streams. + if (!receiveOnly) { + // Fallback for publishers whose offer has no `a=ssrc:N msid:...` + // lines (common with hardware/native WHIP encoders, some OBS + // configurations). Without the msid loop populating streamsMap, + // SMB never gets a `streams` declaration and receivers' SDPs + // end up with no usable msid — so the frontend can't match the + // tile to a participant. Synthesize one stream entry tagged with + // the publisher's endpointId, gathering all video ssrcs from the + // offer (deduped, primary SSRCs of FID groups preferred). + if (streamsMap.size === 0) { + const allVideoSsrcs = (media.ssrcs ?? []).map((s) => + parseInt(`${s.id}`, 10) + ); + const dedupedSsrcs = Array.from(new Set(allVideoSsrcs)); + // If FID groups are present, take the first ssrc of each as + // main and the second as feedback. Otherwise treat each ssrc + // as a primary with no feedback pair. + const fidGroups = (media.ssrcGroups ?? []).filter( + (g) => g.semantics === 'FID' + ); + const sources = + fidGroups.length > 0 + ? fidGroups.map((g) => { + const [mainStr, feedbackStr] = g.ssrcs.split(' '); + return { + main: parseInt(mainStr, 10), + ...(feedbackStr + ? { feedback: parseInt(feedbackStr, 10) } + : {}) + }; + }) + : dedupedSsrcs.map((id) => ({ main: id })); + if (sources.length > 0) { + streamsMap.set(endpointId, { + sources, + id: endpointId, + content: 'video' + }); + } + } + streamsMap.forEach((value) => videoStreams.push(value)); + } + // Only H264 and VP8 are fully supported through the pipeline (codec + // normalization, profile-level-id pinning, FID/RTX handling). VP9 + // used to be in this list but the downstream code never grew + // VP9-specific paths, so it would silently fall through + // misconfigured. + const supportedCodecs = ['VP8', 'H264']; const matchingCodecs = media.rtp?.filter((rtp: RtpCodec) => supportedCodecs.includes(rtp.codec.toUpperCase()) @@ -275,13 +450,42 @@ export class CoreFunctions { endpoint.video = endpoint.video || {}; - const selectedCodec = media.rtp[0]; + // Negotiate a codec both the client offered and SMB advertised. A + // codec the bridge cannot carry leaves receivers with audio and no + // video, so reject explicitly when there is no overlap. + const smbCodecs = smbAdvertisedVideoCodecs(endpoint); + const selectedCodec = selectVideoCodec(media.rtp, smbCodecs); + + if (!selectedCodec) { + throw new Error( + `No video codec in common between the offer and SMB. ` + + `Offered: ${ + media.rtp.map((r) => r.codec).join(', ') || '(none)' + }. ` + + `SMB advertises: ${smbCodecs.join(', ') || '(none)'}.` + ); + } + + // Log both sides; a codec mismatch is otherwise silent. + Log().info( + `[video-codec] endpoint=${endpointId} ` + + `selected=${selectedCodec.codec}@${selectedCodec.payload} ` + + `offered=[${media.rtp.map((r) => r.codec).join(', ')}] ` + + `smb=[${smbCodecs.join(', ') || 'unadvertised'}]` + ); + if (typeof selectedCodec.rate !== 'number') { throw new Error('Selected video codec is missing a valid clockrate'); } + // Always use the offer's actual PT for both WHIP publishers and WHEP + // receivers. whip-mpegts natively offers H264@PT96 so it is unaffected; + // browser WHIP clients offer H264@PT103 and must configure at PT103 so + // SMB knows which PT to forward. WHEP receivers keep their native PT. + const normalizedId = selectedCodec.payload; + const payload = { - id: selectedCodec.payload, + id: normalizedId, name: selectedCodec.codec, clockrate: selectedCodec.rate, parameters: {}, @@ -316,8 +520,67 @@ export class CoreFunctions { uri: ext.uri })); - endpoint.video.ssrcs = - media.ssrcs?.map((ssrc) => Number(ssrc.id)) ?? []; + // Sync payload-types (plural, from allocation, normalized to PT 96) to + // match the actual offer PT so both fields in the configure body agree. + // Applies to all endpoints — WHIP publishers and WHEP receivers alike. + // whip-mpegts already uses PT 96, so no-op for that path. + const payloadTypesArr = endpoint.video['payload-types']; + if (payloadTypesArr && payloadTypesArr.length > 0) { + const mainEntry = payloadTypesArr.find( + (pt) => pt.name.toLowerCase() !== 'rtx' + ); + if (mainEntry) { + mainEntry.id = normalizedId; + } + const rtxEntry = payloadTypesArr.find( + (pt) => pt.name.toLowerCase() === 'rtx' + ); + if (rtxEntry?.parameters?.apt !== undefined) { + rtxEntry.parameters.apt = String(normalizedId); + } + } + + if (!receiveOnly && videoStreams.length > 0) { + // WHIP/camera sender: declare the SSRCs being transmitted and the + // stream so SMB knows what to forward to other endpoints. + endpoint.video.ssrcs = + media.ssrcs?.map((ssrc) => Number(ssrc.id)) ?? []; + endpoint.video.streams = videoStreams; + + // Block all video EGRESS to this publisher. An empty-but-present + // ssrc-whitelist is the one setting SMB reads as "forward nothing" — + // deleting the key instead means last-N, i.e. forward everything. + // + // A WHIP publisher such as whip-mpegts negotiates recvonly on SMB's + // side and builds no video receive path, so any video SMB forwards + // here arrives at a webrtcbin transport with nothing linked + // downstream. That is a fatal GST_FLOW_NOT_LINKED: the whole + // pipeline errors out and the publisher's own outbound video + // freezes. Triggered by any video sender in the conference — one + // already present when the publisher connects, or one joining later. + // + // Ingress is unaffected: the endpoint is still allocated with video + // and still declares its own ssrcs/streams above, so SMB keeps + // receiving this publisher's video and relaying it to subscribers. + // Egress and ingress are independent here. + endpoint.video['ssrc-whitelist'] = []; + } else if (receiveOnly && subscribeToVideo) { + // Keep the pre-allocated receive SSRCs from the allocation (they + // define this endpoint's ssrc-rewrite receive pool). + delete endpoint.video.streams; + const whitelist = subscribeToVideo.ssrcs.slice(0, 2); + if (whitelist.length > 0) { + endpoint.video['ssrc-whitelist'] = whitelist; + } + } else { + // Receive-only WHEP endpoint (ssrc-rewrite mode): keep the pre- + // allocated receive SSRCs from the allocation — SMB needs them to + // set up the ssrc-rewrite forwarding path for this subscriber. + // Only delete 'streams' (this endpoint does not publish video). + // Old forwarder mode deleted both, but ssrc-rewrite requires the + // receive pool to be declared so SMB maps publisher SSRCs to it. + delete endpoint.video.streams; + } } } @@ -460,32 +723,32 @@ export class CoreFunctions { media.ext = audioExts.map((ext) => ({ value: ext.id, uri: ext.uri })); } else if (media.type === 'video') { - const vp8Codec = media.rtp.find( - (rtp: RtpCodec) => rtp.codec.toUpperCase() === 'VP8' + // Same rule as configureEndpointForWhipWhep: the answer decides what + // the publisher encodes and the configure what SMB expects, so the two + // must agree. + const primaryCodec = selectVideoCodec( + media.rtp, + smbAdvertisedVideoCodecs(endpoint) ); - if (vp8Codec) { - const vp8PayloadType = vp8Codec.payload; + + if (primaryCodec) { + const primaryPt = primaryCodec.payload; const rtxFmtp = media.fmtp.find( - (fmtp: Fmtp) => fmtp.config === `apt=${vp8PayloadType}` + (fmtp: Fmtp) => fmtp.config === `apt=${primaryPt}` ); - const vp8RtxPayloadType = rtxFmtp?.payload; + const rtxPt = rtxFmtp?.payload; media.rtp = media.rtp.filter( (rtp: RtpCodec) => - rtp.payload === vp8PayloadType || - rtp.payload === vp8RtxPayloadType + rtp.payload === primaryPt || rtp.payload === rtxPt ); media.fmtp = media.fmtp.filter( - (fmtp: Fmtp) => - fmtp.payload === vp8PayloadType || - fmtp.payload === vp8RtxPayloadType + (fmtp: Fmtp) => fmtp.payload === primaryPt || fmtp.payload === rtxPt ); - media.payloads = [vp8PayloadType, vp8RtxPayloadType] - .filter(Boolean) - .join(' '); + media.payloads = [primaryPt, rtxPt].filter(Boolean).join(' '); media.ext = media.ext?.filter( (ext: RtpHeaderExt) => @@ -494,19 +757,38 @@ export class CoreFunctions { ext.uri === 'urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id' ) ?? []; + // Keep nack (incl. nack pli for keyframe requests), ccm fir, and + // goog-remb. PLI is what lets a WHEP consumer ask the publisher for + // an IDR — without it, a consumer that joins mid-stream or loses + // the reference frame has nothing decodable until the next + // keyframe. media.rtcpFb = media.rtcpFb?.filter( (fb: RtcpFb) => - fb.payload === vp8PayloadType && - (fb.type === 'goog-remb' || fb.type === 'nack') + fb.payload === primaryPt && + (fb.type === 'goog-remb' || + fb.type === 'nack' || + (fb.type === 'ccm' && fb.subtype === 'fir')) + ); + Log().debug( + `[whipwhep-answer] video rtcp-fb negotiated mid=${ + media.mid + } pt=${primaryPt} fb=${JSON.stringify(media.rtcpFb)}` ); media.setup = 'active'; media.direction = media.direction === 'recvonly' ? 'sendonly' : 'recvonly'; media.ssrcGroups = undefined; + // Do not declare a specific a=ssrc: in the WHEP answer. SMB sends + // two SSRCs (main video + RTX); declaring ssrcs[0] risks picking the + // RTX SSRC, which causes Chrome to bind the video track to the repair + // stream (framesDecoded=0) while the actual video arrives on the + // undeclared main SSRC. Without a=ssrc: Chrome accepts all SSRCs on + // this m-line and fires ontrack when real video frames arrive. + media.ssrcs = []; } else { Log().warn( - 'No VP8 codec found in offer video media. Skipping VP8-specific filtering.' + 'No H264 or VP8 codec found in offer video media. Skipping video codec filtering.' ); media.setup = 'active'; media.direction = @@ -539,7 +821,11 @@ export class CoreFunctions { lineId: string, endpointId: string, endpointDescription: SmbEndpointDescription, - answer: string + answer: string, + // Optional pin: when provided, the browser endpoint's video receive + // is gated to packets whose inbound SSRC is in the whitelist. + // Without it, SMB rotates senders through the single video m-line + subscribeToVideo?: { ssrcs: number[]; endpointId: string } ): Promise { if (!endpointDescription) { throw new Error( @@ -572,10 +858,85 @@ export class CoreFunctions { if (endpointDescription.audio.ssrcs.length === 0) { throw new Error( - 'Missing audio ssrcs when handling sdp answer from endpoint' + 'Missing audio ssrcs in SDP answer — answer had no a=ssrc on the ' + + 'audio m-line (mic muted or track removed before negotiation).' ); } + const videoMedia = + parsedAnswer.media.find( + (m) => m.type === 'video' && (m.ssrcs?.length ?? 0) > 0 + ) ?? parsedAnswer.media.find((m) => m.type === 'video'); + if (endpointDescription.video) { + const video = endpointDescription.video; + const videoSsrcs: number[] = []; + video.ssrcs = videoSsrcs; + + if (videoMedia) { + // Extract sender SSRCs from the answer (empty for recvonly/no-camera clients) + if (videoMedia.ssrcs?.length) { + const seen = new Set(); + videoMedia.ssrcs.forEach((ssrc) => { + const id = + typeof ssrc.id === 'string' ? parseInt(ssrc.id, 10) : ssrc.id; + if (!seen.has(id)) { + seen.add(id); + videoSsrcs.push(id); + } + }); + } + + const videoPayloadInfo = this.extractVideoPayloadInfo(videoMedia); + if (videoPayloadInfo) { + endpointDescription.video['payload-type'] = + videoPayloadInfo.payloadType; + endpointDescription.video['rtp-hdrexts'] = + videoPayloadInfo.rtpHdrexts; + } + } + + // Replace the allocate's 'streams' (pre-allocated SSRCs) with the actual + // SSRCs the client is sending. In forwarder mode SMB routes based on the + // SSRC reported in 'streams', so it must match what the browser sends. + // For receive-only (no-camera) clients, streams is empty. + if (videoSsrcs.length > 0) { + // Camera client: build a stream entry from the FID group (main + RTX) + const fidGroup = (videoMedia as any)?.ssrcGroups?.find( + (g: { semantics: string; ssrcs: string }) => g.semantics === 'FID' + ); + let sources: { main: number; feedback?: number }[]; + if (fidGroup) { + const parts = fidGroup.ssrcs.split(' '); + const main = parseInt(parts[0], 10); + const feedback = + parts.length >= 2 ? parseInt(parts[1], 10) : undefined; + sources = [ + feedback !== undefined && Number.isFinite(feedback) + ? { main, feedback } + : { main } + ]; + // Store BOTH main and RTX SSRCs. Receivers pinned to this + // publisher build their ssrc-whitelist from this list — if + // RTX is missing, SMB drops retransmission packets and any + // network jitter freezes the receiver's video. + endpointDescription.video.ssrcs = + feedback !== undefined && Number.isFinite(feedback) + ? [main, feedback] + : [main]; + } else { + sources = [{ main: videoSsrcs[0] }]; + } + const msidEntry = videoMedia?.ssrcs?.find( + (s) => Number(s.id) === sources[0].main && s.attribute === 'msid' + ); + const streamId = msidEntry?.value?.split(' ')[0] ?? 'video'; + video.streams = [{ id: streamId, content: 'video', sources }]; + } else { + // No-camera client: not sending any video + video.streams = []; + } + } + const transport = endpointDescription['bundle-transport']; if (!transport) { throw new Error( @@ -622,6 +983,36 @@ export class CoreFunctions { }; }); + Log().debug( + `[handleAnswer-video] ssrcs=${JSON.stringify( + endpointDescription.video?.ssrcs + )} streams=${JSON.stringify(endpointDescription.video?.streams)}` + ); + + // Apply ssrc-whitelist so SMB only forwards the pinned publisher's + // packets into this browser's single inbound video slot. SMB caps + // the whitelist at 2 SSRCs (main + RTX), so dedupe and slice. + if (subscribeToVideo && endpointDescription.video) { + const whitelist = Array.from(new Set(subscribeToVideo.ssrcs)) + .filter((n) => Number.isFinite(n)) + .slice(0, 2); + if (whitelist.length > 0) { + endpointDescription.video['ssrc-whitelist'] = whitelist; + Log().debug( + `[handleAnswer-pin] browser endpoint=${endpointId} pinned to ` + + `source endpointId=${subscribeToVideo.endpointId} ` + + `whitelist=${JSON.stringify(whitelist)}` + ); + } else { + Log().warn( + `[handleAnswer-pin] browser endpoint=${endpointId} pin requested ` + + `for source endpointId=${subscribeToVideo.endpointId} but ` + + `the source has no stored ssrcs to whitelist with — falling ` + + `back to default rotation` + ); + } + } + return await smb.configureEndpoint( smbServerUrl, lineId, @@ -660,9 +1051,21 @@ export class CoreFunctions { return line.smbConferenceId; } + // SMB's video receive pool size for ssrc-rewrite endpoints. Each + // endpoint in this conference gets `last-n + 2` simultaneous video + // slots (capped at 16 server-side). Default 9 → 11 slots, which is + // generous enough for our typical conferences while leaving SMB's + // simulcast headroom intact. Env-tunable for ops without a rebuild. + // Required for the WHEP single-source pin to work — without it, + // ssrc-rewrite receivers get zero slots and SMB falls back to last-N + // forwarding (which is what the dynamic-source bug looked like). + const parsedLastN = parseInt(process.env.SMB_CONFERENCE_LAST_N ?? '9', 10); + const lastN = + Number.isFinite(parsedLastN) && parsedLastN >= 1 ? parsedLastN : 9; const newConferenceId = await smb.allocateConference( smbServerUrl, - smbServerApiKey + smbServerApiKey, + lastN ); if ( @@ -704,7 +1107,13 @@ export class CoreFunctions { const allLinesResponse = await Promise.all( production.lines.map( - async ({ name, id, smbConferenceId, programOutputLine }) => { + async ({ + name, + id, + smbConferenceId, + programOutputLine, + videoEnabled + }) => { const participants = await this.productionManager.getUsersForLine( stringifiedProdId, id @@ -715,7 +1124,8 @@ export class CoreFunctions { id, smbConferenceId, participants, - programOutputLine: programOutputLine ?? false + programOutputLine: programOutputLine ?? false, + videoEnabled: videoEnabled ?? false } as LineResponse; } ) @@ -733,4 +1143,68 @@ export class CoreFunctions { throw new Error(`${value} has incorrect type`); } } + + private extractVideoPayloadInfo(media: MediaDescription): { + payloadType: { + id: number; + name: string; + clockrate: number; + parameters: Record; + 'rtcp-fbs': { type: string; subtype?: string }[]; + }; + rtpHdrexts: { id: number; uri: string }[]; + } | null { + // Match the H264-preferred selection used in configureEndpointForWhipWhep. + // Previously this function included VP9 and picked + // matchingRtp[0], so a browser answer listing VP8 before H264 would set + // the receiver's payload-type to VP8's PT while SMB expected H264 — the + // exact PT mismatch addVideoMid normalization was preventing. + if (!media.rtp?.length) return null; + const selectedCodec = + media.rtp.find((rtp: RtpCodec) => rtp.codec.toUpperCase() === 'H264') ?? + media.rtp.find((rtp: RtpCodec) => rtp.codec.toUpperCase() === 'VP8'); + if (!selectedCodec) return null; + if (typeof selectedCodec.rate !== 'number') return null; + + const fmtp = media.fmtp?.find( + (f: Fmtp) => f.payload === selectedCodec.payload + ); + const parameters: Record = fmtp?.config + ? Object.fromEntries( + fmtp.config.split(';').map((kv) => { + const [key, val] = kv.trim().split('='); + return [key, val ?? '']; + }) + ) + : {}; + + const rtcpFbs = ( + media.rtcpFb?.filter( + (f: RtcpFb) => f.payload === selectedCodec.payload + ) ?? [] + ).map((fb: RtcpFb) => ({ + type: fb.type, + subtype: fb.subtype ?? undefined + })); + + const allowedExts = [ + 'http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time', + 'urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id' + ]; + const rtpHdrexts = ( + media.ext?.filter((ext: RtpHeaderExt) => allowedExts.includes(ext.uri)) ?? + [] + ).map((ext: RtpHeaderExt) => ({ id: ext.value, uri: ext.uri })); + + return { + payloadType: { + id: selectedCodec.payload, + name: selectedCodec.codec, + clockrate: selectedCodec.rate, + parameters, + 'rtcp-fbs': rtcpFbs + }, + rtpHdrexts + }; + } } diff --git a/src/api_productions_integration.test.ts b/src/api_productions_integration.test.ts index 985b68e..dfeb5ea 100644 --- a/src/api_productions_integration.test.ts +++ b/src/api_productions_integration.test.ts @@ -186,6 +186,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { conferenceId, 'ep-001', true, // audio + false, // video false, // data true, // iceControlling 'ssrc-rewrite', @@ -213,6 +214,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { conferenceId, 'ep-002', false, // audio + false, // video true, // data false, 'forwarder', @@ -233,6 +235,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-003', true, false, + false, true, 'ssrc-rewrite', 60 @@ -262,6 +265,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { conferenceId, endpointId, true, + false, true, true, 'ssrc-rewrite', @@ -314,6 +318,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-a', true, false, + false, true, 'ssrc-rewrite', 60 @@ -327,6 +332,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-b', true, false, + false, true, 'ssrc-rewrite', 60 @@ -363,6 +369,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-1', true, false, + false, true, 'ssrc-rewrite', 60, @@ -374,6 +381,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-2', true, false, + false, true, 'ssrc-rewrite', 60, @@ -395,6 +403,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-1', true, false, + false, true, 'ssrc-rewrite', 60, @@ -416,6 +425,7 @@ describe('MockSmbProtocol Integration with CoreFunctions', () => { 'ep-1', true, false, + false, true, 'ssrc-rewrite', 60, diff --git a/src/api_productions_video_source.test.ts b/src/api_productions_video_source.test.ts new file mode 100644 index 0000000..28df844 --- /dev/null +++ b/src/api_productions_video_source.test.ts @@ -0,0 +1,375 @@ +// Tests for PATCH /session/:sessionId/video-source: +// a source swap must force a fresh keyframe for the newly pinned ssrc so the +// receiver's decoder recovers immediately instead of freezing on the previous +// publisher's last frame. + +jest.mock('./log', () => ({ + Log: () => ({ + error: jest.fn(), + warn: jest.fn(), + info: jest.fn(), + debug: jest.fn() + }) +})); + +// Capture every SMB call so we can assert ordering and the forced keyframe. +type SmbCall = { + kind: 'reconfigure' | 'requestKeyframe'; + endpointId: string; + ssrcWhitelist: number[] | undefined; +}; +const smbCalls: SmbCall[] = []; + +// When set, the mocked reconfigureEndpoint rejects with it instead of +// succeeding. Lets a test drive the SMB-rejection paths. +let mockReconfigureError: Error | null = null; + +// requireActual keeps the real SmbEndpointActionError — the route checks the +// rejection with `instanceof`, so a factory returning only SmbProtocol would +// leave that check comparing against undefined. +jest.mock('./smb', () => ({ + ...jest.requireActual('./smb'), + SmbProtocol: jest.fn().mockImplementation(() => ({ + reconfigureEndpoint: jest + .fn() + .mockImplementation( + async (_url: string, _conf: string, endpointId: string, desc: any) => { + if (mockReconfigureError) { + throw mockReconfigureError; + } + smbCalls.push({ + kind: 'reconfigure', + endpointId, + ssrcWhitelist: desc?.video?.['ssrc-whitelist'] + }); + } + ), + requestKeyframe: jest + .fn() + .mockImplementation( + async (_url: string, _conf: string, endpointId: string, desc: any) => { + smbCalls.push({ + kind: 'requestKeyframe', + endpointId, + ssrcWhitelist: desc?.video?.['ssrc-whitelist'] + }); + } + ) + })) +})); + +import api from './api'; +import { SmbEndpointActionError } from './smb'; + +const RECEIVER_ENDPOINT = 'receiver-ep-1'; +const SOURCE_A_SSRCS = [1111, 2222]; +const SOURCE_B_SSRCS = [3333, 4444]; + +function makeReceiverSession() { + return { + _id: 'receiver-1', + name: 'receiver', + productionId: '1', + lineId: '1', + isActive: true, + isExpired: false, + isWhip: false, + hasVideo: false, + endpointId: RECEIVER_ENDPOINT, + pinnedVideoSessionId: undefined as string | undefined, + sessionDescription: { + audio: { ssrcs: [9000], 'payload-type': {}, 'rtp-hdrexts': [] }, + video: { ssrcs: [], 'payload-type': {}, 'rtp-hdrexts': [] } + } + }; +} + +function makeSourceSession(id: string, ssrcs: number[]) { + return { + _id: id, + name: id, + productionId: '1', + lineId: '1', + isActive: true, + isExpired: false, + isWhip: true, + hasVideo: true, + endpointId: `${id}-ep`, + sessionDescription: { + audio: { ssrcs: [8000], 'payload-type': {}, 'rtp-hdrexts': [] }, + video: { ssrcs, 'payload-type': {}, 'rtp-hdrexts': [] } + } + }; +} + +let receiverSession: ReturnType; +const sessions: Record = {}; + +const mockDbManager: any = { + connect: jest.fn().mockResolvedValue(undefined), + getSession: jest + .fn() + .mockImplementation(async (id: string) => sessions[id] ?? null), + deleteUserSession: jest.fn().mockResolvedValue(true) +}; + +const mockProductionManager: any = { + checkUserStatus: jest.fn().mockResolvedValue(undefined), + requireProduction: jest + .fn() + .mockResolvedValue({ _id: 1, name: 'prod-1', lines: [{ id: '1' }] }), + requireLine: jest + .fn() + .mockReturnValue({ id: '1', smbConferenceId: 'smb-conf-1' }), + updateSessionVideoPin: jest + .fn() + .mockImplementation( + async (sessionId: string, _desc: any, pinned: string | null) => { + if (sessions[sessionId]) { + sessions[sessionId].pinnedVideoSessionId = pinned ?? undefined; + } + } + ), + // Leave-reconcile dependencies (DELETE /session/:sessionId). + clearWhepSourceIfPinned: jest.fn().mockResolvedValue(undefined), + getReceiversPinnedToSession: jest + .fn() + .mockImplementation(async (leaverId: string) => + Object.values(sessions).filter( + (s: any) => s._id !== leaverId && s.pinnedVideoSessionId === leaverId + ) + ), + getProduction: jest.fn().mockResolvedValue({ + _id: 1, + name: 'prod-1', + lines: [{ id: '1', smbConferenceId: 'smb-conf-1' }] + }), + removeUserSession: jest.fn(), + emit: jest.fn() +}; + +const mockIngestManager: any = { + load: jest.fn().mockResolvedValue(undefined), + startPolling: jest.fn() +}; + +const mockCoreFunctions: any = { + getAllLinesResponse: jest.fn().mockReturnValue([]) +}; + +describe('PATCH /session/:sessionId/video-source — forced keyframe', () => { + let server: any; + let setIntervalSpy: jest.SpyInstance; + + beforeAll(async () => { + setIntervalSpy = jest + .spyOn(global, 'setInterval') + .mockImplementation(jest.fn() as any); + server = await api({ + title: 'video-source test', + smbServerBaseUrl: 'http://localhost', + endpointIdleTimeout: '60', + publicHost: 'https://example.com', + dbManager: mockDbManager, + productionManager: mockProductionManager, + ingestManager: mockIngestManager, + coreFunctions: mockCoreFunctions + }); + }); + + afterAll(async () => { + await server.close(); + setIntervalSpy.mockRestore(); + }); + + beforeEach(() => { + smbCalls.length = 0; + mockReconfigureError = null; + receiverSession = makeReceiverSession(); + for (const key of Object.keys(sessions)) delete sessions[key]; + sessions['receiver-1'] = receiverSession; + sessions['source-a'] = makeSourceSession('source-a', SOURCE_A_SSRCS); + sessions['source-b'] = makeSourceSession('source-b', SOURCE_B_SSRCS); + }); + + async function pin(pinnedSessionId: string | null) { + return server.inject({ + method: 'PATCH', + url: '/api/v1/session/receiver-1/video-source', + body: { pinnedSessionId } + }); + } + + test('forces a keyframe for the new ssrc on initial pin', async () => { + const res = await pin('source-a'); + expect(res.statusCode).toBe(200); + + const keyframeCalls = smbCalls.filter((c) => c.kind === 'requestKeyframe'); + expect(keyframeCalls).toHaveLength(1); + expect(keyframeCalls[0].endpointId).toBe(RECEIVER_ENDPOINT); + expect(keyframeCalls[0].ssrcWhitelist).toEqual(SOURCE_A_SSRCS); + }); + + // An endpoint is allocated before it is configured, and a client that pins a + // publisher the moment it appears can land in between. SMB rejects the + // reconfigure with a 400; answering 500 would read as a real failure and the + // client would not retry, leaving the pin silently abandoned. + test('returns 425 when the session endpoint is not configured on SMB yet', async () => { + mockReconfigureError = new SmbEndpointActionError( + 'reconfigure', + 400, + JSON.stringify({ + message: + "Can't reconfigure audio because it was not configured in first place", + status_code: 400 + }) + ); + + // This suite does not reset mock call history between tests, so compare + // against the count taken just before the request rather than an absolute. + const pinWrites = () => + (mockProductionManager.updateSessionVideoPin as jest.Mock).mock.calls + .length; + const pinWritesBefore = pinWrites(); + + const res = await pin('source-a'); + + expect(res.statusCode).toBe(425); + expect(JSON.parse(res.body).message).toMatch(/retry shortly/i); + // Nothing must be persisted or forced when the pin did not take effect. + expect(smbCalls.filter((c) => c.kind === 'requestKeyframe')).toHaveLength( + 0 + ); + expect(pinWrites()).toBe(pinWritesBefore); + }); + + test('still fails loudly for an unrelated SMB rejection', async () => { + mockReconfigureError = new SmbEndpointActionError( + 'reconfigure', + 400, + JSON.stringify({ message: 'Some other bad request', status_code: 400 }) + ); + + const res = await pin('source-a'); + + expect(res.statusCode).toBe(500); + }); + + test('forces a keyframe for the NEW source when the pin is swapped', async () => { + // Receiver already pinned to source-a (e.g. Maj), now swaps to source-b + // (e.g. Elsa auto-pinned after Maj leaves). + receiverSession.pinnedVideoSessionId = 'source-a'; + + const res = await pin('source-b'); + expect(res.statusCode).toBe(200); + + const keyframeCalls = smbCalls.filter((c) => c.kind === 'requestKeyframe'); + expect(keyframeCalls).toHaveLength(1); + expect(keyframeCalls[0].ssrcWhitelist).toEqual(SOURCE_B_SSRCS); + + // The keyframe must be requested AFTER the whitelist reconfigure lands. + const reconfigureIdx = smbCalls.findIndex((c) => c.kind === 'reconfigure'); + const keyframeIdx = smbCalls.findIndex((c) => c.kind === 'requestKeyframe'); + expect(reconfigureIdx).toBeGreaterThanOrEqual(0); + expect(keyframeIdx).toBeGreaterThan(reconfigureIdx); + }); + + test('does NOT force a keyframe for a no-op re-pin to the same source', async () => { + receiverSession.pinnedVideoSessionId = 'source-a'; + + const res = await pin('source-a'); + expect(res.statusCode).toBe(200); + + expect(smbCalls.filter((c) => c.kind === 'requestKeyframe')).toHaveLength( + 0 + ); + }); + + test('does NOT force a keyframe when clearing the pin', async () => { + receiverSession.pinnedVideoSessionId = 'source-a'; + + const res = await pin(null); + expect(res.statusCode).toBe(200); + + expect(smbCalls.filter((c) => c.kind === 'requestKeyframe')).toHaveLength( + 0 + ); + }); +}); + +describe('DELETE /session/:sessionId — reconcile dangling video pins (leave fix)', () => { + let server: any; + let setIntervalSpy: jest.SpyInstance; + + beforeAll(async () => { + setIntervalSpy = jest + .spyOn(global, 'setInterval') + .mockImplementation(jest.fn() as any); + server = await api({ + title: 'video-source leave test', + smbServerBaseUrl: 'http://localhost', + endpointIdleTimeout: '60', + publicHost: 'https://example.com', + dbManager: mockDbManager, + productionManager: mockProductionManager, + ingestManager: mockIngestManager, + coreFunctions: mockCoreFunctions + }); + }); + + afterAll(async () => { + await server.close(); + setIntervalSpy.mockRestore(); + }); + + beforeEach(() => { + smbCalls.length = 0; + mockReconfigureError = null; + receiverSession = makeReceiverSession(); + for (const key of Object.keys(sessions)) delete sessions[key]; + sessions['receiver-1'] = receiverSession; + sessions['source-a'] = makeSourceSession('source-a', SOURCE_A_SSRCS); + sessions['source-b'] = makeSourceSession('source-b', SOURCE_B_SSRCS); + }); + + async function leave(sessionId: string) { + return server.inject({ + method: 'DELETE', + url: `/api/v1/session/${sessionId}` + }); + } + + test('clears a receiver whitelist when the pinned source leaves', async () => { + // receiver-1 is pinned to source-a, and its endpoint description still + // carries source-a's SSRCs as the ssrc-whitelist (set by an earlier pin). + receiverSession.pinnedVideoSessionId = 'source-a'; + (receiverSession.sessionDescription.video as any)['ssrc-whitelist'] = + SOURCE_A_SSRCS; + + const res = await leave('source-a'); + expect(res.statusCode).toBe(200); + + // The receiver must be reconfigured with the whitelist REMOVED (key + // deleted -> last-N fallback), never an empty-but-enabled whitelist. + const reconfigures = smbCalls.filter((c) => c.kind === 'reconfigure'); + expect(reconfigures).toHaveLength(1); + expect(reconfigures[0].endpointId).toBe(RECEIVER_ENDPOINT); + expect(reconfigures[0].ssrcWhitelist).toBeUndefined(); + + // The stored pin is cleared so the client's auto-pin effect re-pins fresh. + expect(receiverSession.pinnedVideoSessionId).toBeUndefined(); + }); + + test('does NOT touch receivers that pinned someone else', async () => { + // receiver-1 is pinned to source-b; source-a (which nobody pinned) leaves. + receiverSession.pinnedVideoSessionId = 'source-b'; + (receiverSession.sessionDescription.video as any)['ssrc-whitelist'] = + SOURCE_B_SSRCS; + + const res = await leave('source-a'); + expect(res.statusCode).toBe(200); + + expect(smbCalls.filter((c) => c.kind === 'reconfigure')).toHaveLength(0); + expect(receiverSession.pinnedVideoSessionId).toBe('source-b'); + }); +}); diff --git a/src/api_whep.ts b/src/api_whep.ts index 0057d98..9f2d631 100644 --- a/src/api_whep.ts +++ b/src/api_whep.ts @@ -141,6 +141,53 @@ export const apiWhep: FastifyPluginCallback = ( const sessionId = uuidv4(); const endpointId = uuidv4(); + const offerHasVideo = sdpOffer.media.some((m) => m.type === 'video'); + + let subscribeToVideo: + | { streams: any[]; ssrcs: number[]; endpointId: string } + | undefined; + try { + const productionIdNum = parseInt(productionId, 10); + if (!Number.isNaN(productionIdNum)) { + const production = await productionManager.getProduction( + productionIdNum + ); + const line = production?.lines.find((l) => l.id === lineId); + const pinnedSessionId = line?.whepSourceSessionId ?? null; + if (pinnedSessionId) { + const sourceSession = await opts.dbManager.getSession( + pinnedSessionId + ); + const sourceVideo: any = sourceSession?.sessionDescription?.video; + const sourceEndpointId = sourceSession?.endpointId; + const streams: any[] = Array.isArray(sourceVideo?.streams) + ? sourceVideo.streams + : []; + const ssrcs: number[] = Array.isArray(sourceVideo?.ssrcs) + ? sourceVideo.ssrcs + : []; + if ( + sourceEndpointId && + (streams.length > 0 || ssrcs.length > 0) + ) { + subscribeToVideo = { + streams, + ssrcs, + endpointId: sourceEndpointId + }; + } else { + Log().warn( + `[whep-pin] line=${lineId} pinned sessionId=${pinnedSessionId} has no usable sessionDescription.video — falling back to forward-all` + ); + } + } + } + } catch (pinErr) { + Log().warn( + `[whep-pin] failed to resolve pinned source, falling back to forward-all: ${pinErr}` + ); + } + // Create conference and endpoint in SMB const smbConferenceId = await coreFunctions.createConferenceForLine( smb, @@ -150,7 +197,6 @@ export const apiWhep: FastifyPluginCallback = ( lineId ); - // Allocate endpoint with audio support const endpoint = await coreFunctions.createEndpoint( smb, smbServerUrl, @@ -158,10 +204,16 @@ export const apiWhep: FastifyPluginCallback = ( smbConferenceId, endpointId, true, // audio + offerHasVideo, // video false, // no data channel needed for WHEP true, // iceControlling - 'ssrc-rewrite', // relayType - parseInt(opts.endpointIdleTimeout, 10) + 'ssrc-rewrite', // audio relay type + parseInt(opts.endpointIdleTimeout, 10), + 'ssrc-rewrite' + ); + + Log().debug( + `[whep-alloc] video.ssrcs=${JSON.stringify(endpoint.video?.ssrcs)}` ); await coreFunctions.configureEndpointForWhipWhep( @@ -171,7 +223,9 @@ export const apiWhep: FastifyPluginCallback = ( smbServerUrl, smbServerApiKey, smbConferenceId, - endpointId + endpointId, + true, // receiveOnly: WHEP is receive-only + subscribeToVideo ); const sdpAnswer = await coreFunctions.createWhipWhepAnswer( @@ -215,7 +269,9 @@ export const apiWhep: FastifyPluginCallback = ( lineId, sessionId, username, - true + true, // isWhip — kept for backwards compat with consumers + true, // isWhepReceiver — distinguishes egress recipients from WHIP publishers + false // hasVideo — WHEP is receive-only by spec, never publishes ); // Update user endpoint info and store a stable smbPresenceKey @@ -281,6 +337,8 @@ export const apiWhep: FastifyPluginCallback = ( return; } + await productionManager.clearWhepSourceIfPinned(sessionId); + await opts.dbManager.deleteUserSession(sessionId); productionManager.removeUserSession(sessionId); productionManager.emit('users:change'); diff --git a/src/api_whip.test.ts b/src/api_whip.test.ts index 5dcb291..04e88be 100644 --- a/src/api_whip.test.ts +++ b/src/api_whip.test.ts @@ -30,6 +30,9 @@ const mockProductionManager = { deleteProduction: jest.fn().mockResolvedValue(true), getUser: jest.fn().mockResolvedValue(undefined), requireLine: jest.fn().mockResolvedValue({}), + clearWhepSourceIfPinned: jest.fn().mockResolvedValue(undefined), + updateSessionHasVideo: jest.fn().mockResolvedValue(undefined), + setLineWhepSource: jest.fn().mockResolvedValue(undefined), once: jest.fn(), emit: jest.fn() } as any; @@ -139,6 +142,28 @@ describe('apiWhip', () => { jest.clearAllMocks(); }); + /** + * WHIP ingest uses 'ssrc-rewrite' for video, like every other endpoint in the + * system. It used 'forwarder' historically, on a rationale measured for WHEP + * consumers that never applied to a publisher. + */ + describe('video relay type', () => { + const videoRelayArg = () => + (coreFunctions.createEndpoint as jest.Mock).mock.calls[0][11]; + + it("requests 'ssrc-rewrite' video relay for a WHIP publisher", async () => { + const fastify = await createTestServer(); + await fastify.inject({ + method: 'POST', + url: '/whip/prod1/line1/testuser', + headers: { 'content-type': 'application/sdp' }, + payload: + 'v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\nm=audio 0 RTP/AVP 0\r\na=mid:0\r\n' + }); + expect(videoRelayArg()).toBe('ssrc-rewrite'); + }); + }); + describe('POST /whip/:productionId/:lineId/:username', () => { it('should return 201 with SDP answer and proper headers', async () => { const fastify = await createTestServer(); @@ -388,3 +413,103 @@ describe('apiWhip', () => { }); }); }); + +/** + * hasVideo advertises a session as a pin source, so it must mean "has sending + * video SSRCs persisted", not "the offer had a video m-line". + */ +describe('apiWhip hasVideo', () => { + const sdp = (lines: string[]) => + ['v=0', 'o=- 0 0 IN IP4 127.0.0.1', ...lines].join('\r\n') + '\r\n'; + + const AUDIO = ['m=audio 9 RTP/AVP 111', 'a=mid:0']; + + const post = async (payload: string) => { + const fastify = await createTestServer(); + return fastify.inject({ + method: 'POST', + url: '/whip/prod1/line1/testuser', + headers: { 'content-type': 'application/sdp' }, + payload + }); + }; + + const hasVideoCalls = () => + (mockProductionManager.updateSessionHasVideo as jest.Mock).mock.calls; + const storedEndpoint = () => + (mockProductionManager.updateUserEndpoint as jest.Mock).mock.calls[0]?.[2]; + + beforeEach(() => { + jest.clearAllMocks(); + // A fresh endpoint per call: the shared mock otherwise hands every test + // the same object, so SSRCs stamped by one test leak into the next. + (coreFunctions.createEndpoint as jest.Mock).mockImplementation( + async () => ({ + 'bundle-transport': { + 'rtcp-mux': true, + ice: { ufrag: 'test-ufrag', pwd: 'test-pwd', candidates: [] }, + dtls: { fingerprint: 'sha-256 FAKEFINGERPRINT', setup: 'actpass' } + } + }) + ); + // Echo the offer's mids: the route 406s when a mid is missing from the + // answer, and the shared mock answers audio-only. + (coreFunctions.createWhipWhepAnswer as jest.Mock).mockImplementation( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + async (offer: any) => + [ + 'v=0', + 'o=- 0 0 IN IP4 127.0.0.1', + // eslint-disable-next-line @typescript-eslint/no-explicit-any + ...offer.media.map( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (m: any) => `m=${m.type} 9 RTP/AVP 96\r\na=mid:${m.mid}` + ) + ].join('\r\n') + '\r\n' + ); + }); + + it('advertises a publisher whose offer carries an FID group', async () => { + await post( + sdp([ + ...AUDIO, + 'm=video 9 RTP/AVP 96', + 'a=mid:1', + 'a=ssrc-group:FID 111111 222222', + 'a=ssrc:111111 cname:probe', + 'a=ssrc:222222 cname:probe' + ]) + ); + + expect(hasVideoCalls()).toEqual([['mock-session-id', true]]); + // Both main and RTX: without the RTX SSRC SMB drops retransmissions. + expect(storedEndpoint()?.video?.ssrcs).toEqual([111111, 222222]); + }); + + it('advertises a publisher offering a bare a=ssrc with no FID group', async () => { + await post( + sdp([ + ...AUDIO, + 'm=video 9 RTP/AVP 96', + 'a=mid:1', + 'a=ssrc:333333 cname:probe' + ]) + ); + + expect(hasVideoCalls()).toEqual([['mock-session-id', true]]); + expect(storedEndpoint()?.video?.ssrcs).toEqual([333333]); + }); + + it('does NOT advertise a video m-line whose SSRCs cannot be parsed', async () => { + await post(sdp([...AUDIO, 'm=video 9 RTP/AVP 96', 'a=mid:1'])); + + expect(hasVideoCalls()).toEqual([]); + expect(storedEndpoint()?.video?.ssrcs).toBeUndefined(); + }); + + it('does NOT advertise an audio-only publisher', async () => { + await post(sdp(AUDIO)); + + expect(hasVideoCalls()).toEqual([]); + }); +}); diff --git a/src/api_whip.ts b/src/api_whip.ts index 52d82ed..6fe1181 100644 --- a/src/api_whip.ts +++ b/src/api_whip.ts @@ -5,7 +5,12 @@ import sdpTransform, { parse } from 'sdp-transform'; import { v4 as uuidv4 } from 'uuid'; import { CoreFunctions } from './api_productions_core_functions'; import { Log } from './log'; -import { Line, WhipWhepRequest, WhipWhepResponse } from './models'; +import { + Line, + SmbEndpointDescription, + WhipWhepRequest, + WhipWhepResponse +} from './models'; import { ProductionManager } from './production_manager'; import { ISmbProtocol, SmbProtocol } from './smb'; import { getIceServers } from './utils'; @@ -137,6 +142,8 @@ export const apiWhip: FastifyPluginCallback = ( const sdpOffer = parse(request.body); + const offerHasVideo = sdpOffer.media.some((m) => m.type === 'video'); + // Create a unique session ID for this WHIP connection const sessionId = uuidv4(); const endpointId = uuidv4(); @@ -150,7 +157,9 @@ export const apiWhip: FastifyPluginCallback = ( lineId ); - // Allocate endpoint with audio support + // Allocate endpoint with audio (and video, when the offer includes a + // video m= section). SMB requires video to be allocated before a + // configure call can send a video block const endpoint = await coreFunctions.createEndpoint( smb, smbServerUrl, @@ -158,10 +167,12 @@ export const apiWhip: FastifyPluginCallback = ( smbConferenceId, endpointId, true, // audio + offerHasVideo, // video false, // no data channel needed for WHIP true, // iceControlling - 'ssrc-rewrite', // relayType - parseInt(opts.endpointIdleTimeout, 10) + 'ssrc-rewrite', // audio relay type + parseInt(opts.endpointIdleTimeout, 10), + 'ssrc-rewrite' ); await coreFunctions.configureEndpointForWhipWhep( @@ -174,6 +185,27 @@ export const apiWhip: FastifyPluginCallback = ( endpointId ); + if (offerHasVideo) { + const videoMedia = sdpOffer.media.find((m) => m.type === 'video'); + const fidGroup = videoMedia?.ssrcGroups?.find( + (g) => g.semantics === 'FID' + ); + const ssrcs: number[] = []; + if (fidGroup) { + for (const part of fidGroup.ssrcs.split(' ')) { + const n = parseInt(part, 10); + if (Number.isFinite(n)) ssrcs.push(n); + } + } else { + const fallback = Number(videoMedia?.ssrcs?.[0]?.id); + if (Number.isFinite(fallback)) ssrcs.push(fallback); + } + if (ssrcs.length > 0) { + if (!endpoint.video) endpoint.video = {}; + endpoint.video.ssrcs = ssrcs; + } + } + const sdpAnswer = await coreFunctions.createWhipWhepAnswer( sdpOffer, endpoint @@ -206,26 +238,42 @@ export const apiWhip: FastifyPluginCallback = ( } // Create user session in production manager (await to guarantee DB state) - Log().info( + Log().debug( `Creating WHIP user session - username: ${username}, sessionId: ${sessionId}, production: ${productionId}, line: ${lineId}` ); - await productionManager.createUserSession( smbConferenceId, productionId, lineId, sessionId, username, - true + true, // isWhip + false, // isWhepReceiver + false // hasVideo flipped below once video.ssrcs is persisted ); - // Update user endpoint information await productionManager.updateUserEndpoint( sessionId, endpointId, endpoint ); + // Now that video.ssrcs is persisted, flip hasVideo so receivers' + // auto-pin lookup finds this publisher with a usable whitelist. + // Bound to the SSRCs actually persisted, the same rule the browser + // path uses: a video m-line whose SSRCs do not parse would otherwise + // be advertised as a pin source and 425 every pin. + const publishedVideoSsrcs = endpoint.video?.ssrcs ?? []; + if (publishedVideoSsrcs.length > 0) { + await productionManager.updateSessionHasVideo(sessionId, true); + } else if (offerHasVideo) { + Log().warn( + `WHIP offer for session=${sessionId} has a video m-line but no ` + + `usable SSRCs (no FID group, no a=ssrc): not advertising it as ` + + `a pin source, so its video cannot be pinned by receivers.` + ); + } + // Create the Location URL for the WHIP resource // Location URL can be relative to Request URL, so this is OK. const locationUrl = `/api/v1/whip/${productionId}/${lineId}/${sessionId}`; @@ -282,7 +330,56 @@ export const apiWhip: FastifyPluginCallback = ( return; } - // Remove the user session + await productionManager.clearWhepSourceIfPinned(sessionId); + + // Reconcile receivers pinned to this departing WHIP publisher: strip + // their stale ssrc-whitelist so their video does not freeze. Mirrors + // the reconciliation in the PATCH /session/:sessionId DELETE path. + try { + const affected = await productionManager.getReceiversPinnedToSession( + sessionId + ); + if (affected.length > 0) { + const production = await productionManager.getProduction( + parseInt(affected[0].productionId, 10) + ); + const line = production?.lines.find( + (l) => l.id === affected[0].lineId + ); + if (line) { + await Promise.all( + affected.map(async (receiver) => { + const receiverId = (receiver as any)._id?.toString?.(); + const endpointId = receiver.endpointId; + const endpointDescription = receiver.sessionDescription; + if (!receiverId || !endpointId || !endpointDescription) + return; + const updatedDescription: SmbEndpointDescription = JSON.parse( + JSON.stringify(endpointDescription) + ); + if (updatedDescription.video) { + delete updatedDescription.video['ssrc-whitelist']; + } + await smb.reconfigureEndpoint( + smbServerUrl, + line.smbConferenceId, + endpointId, + updatedDescription, + smbServerApiKey + ); + await productionManager.updateSessionVideoPin( + receiverId, + updatedDescription, + null + ); + }) + ); + } + } + } catch { + // Never let pin reconciliation block the WHIP delete itself. + } + await opts.dbManager.deleteUserSession(sessionId); productionManager.removeUserSession(sessionId); productionManager.emit('users:change'); diff --git a/src/connection.ts b/src/connection.ts index 88fa3c1..de665d8 100644 --- a/src/connection.ts +++ b/src/connection.ts @@ -3,8 +3,13 @@ import { SessionDescription } from 'sdp-transform'; import { AudioSmbPayloadParameters, MediaDescriptionBase, - SfuEndpointDescription + SfuEndpointDescription, + VideoSmbPayloadType } from './sfu/interface'; +import { + NORMALIZED_VIDEO_PT_MAIN, + NORMALIZED_VIDEO_PT_RTX +} from './sfu/constants'; import { MediaStreamsInfo } from './media_streams_info'; import { Log } from './log'; @@ -77,6 +82,12 @@ export class Connection { ); msidSemanticToken = `${mslabels.join(' ')}`; } + if (this.mediaStreams.video?.ssrcs.length) { + const videoMslabels = this.mediaStreams.video.ssrcs.map( + (element) => element.mslabel + ); + msidSemanticToken = `${msidSemanticToken} ${videoMslabels.join(' ')}`; + } } offer.msidSemantic = { @@ -164,6 +175,125 @@ export class Connection { return result; } + protected addVideoMid(offer: SessionDescription) { + if (!this.endpointDescription?.video) return; + if (!this.mediaStreams?.video) return; + + const video = this.endpointDescription.video; + + const rawPayloadTypes: VideoSmbPayloadType[] = + video['payload-types'] ?? + (video['payload-type'] ? [video['payload-type']] : []); + + if (!rawPayloadTypes.length) return; + + // Prefer H264; fall back to VP8 for older SMB deployments. + // Restrict to a single codec to prevent Chrome from picking VP9 + // which would cause a PT mismatch with SMB and dropped video packets. + const h264Raw = rawPayloadTypes.find( + (pt) => pt.name.toUpperCase() === 'H264' + ); + const vp8Raw = rawPayloadTypes.find( + (pt) => pt.name.toUpperCase() === 'VP8' + ); + const preferredCodec = h264Raw ?? vp8Raw; + if (!preferredCodec) return; + + const rtxRaw = rawPayloadTypes.find( + (pt) => pt.name.toLowerCase() === 'rtx' + ); + + const mainPt = NORMALIZED_VIDEO_PT_MAIN; + const rtxPt = NORMALIZED_VIDEO_PT_RTX; + + const payloadTypes: VideoSmbPayloadType[] = [ + { ...preferredCodec, id: mainPt }, + ...(rtxRaw + ? [ + { + ...rtxRaw, + id: rtxPt, + parameters: { ...rtxRaw.parameters, apt: String(mainPt) } + } + ] + : []) + ]; + + Log().debug( + `[addVideoMid] smb rtcp-fbs for ${ + preferredCodec.name + } pt=${mainPt} fbs=${JSON.stringify(preferredCodec['rtcp-fbs'] ?? [])}` + ); + + // Helper that builds one video m-line with the shared codec/ext block. + // Caller decides what ssrcs (if any) to put on it. + const buildVideoDescription = () => { + const md = this.makeMediaDescription('video'); + md.payloads = payloadTypes.map((pt) => pt.id).join(' '); + md.rtp = payloadTypes.map((pt) => ({ + payload: pt.id, + codec: pt.name, + rate: pt.clockrate + })); + md.fmtp = payloadTypes + .filter((pt) => pt.parameters && Object.keys(pt.parameters).length > 0) + .map((pt) => ({ + payload: pt.id, + config: Object.entries(pt.parameters) + .map(([k, v]) => (v ? `${k}=${v}` : k)) + .join(';') + })); + md.rtcpFb = payloadTypes + .filter((pt) => pt['rtcp-fbs']?.length) + .flatMap((pt) => + pt['rtcp-fbs'].map((fb) => ({ + payload: pt.id, + type: fb.type, + subtype: fb.subtype ?? '' + })) + ); + if (video['rtp-hdrexts']?.length) { + md.ext = video['rtp-hdrexts'].map((ext) => ({ + value: ext.id, + uri: ext.uri + })); + } + return md; + }; + + const videoSsrcs = this.mediaStreams.video.ssrcs; + + if (videoSsrcs.length === 0) { + offer.media.push(buildVideoDescription()); + return; + } + + for (const element of videoSsrcs) { + const md = buildVideoDescription(); + md.ssrcs.push({ + id: Number(element.ssrc), + attribute: 'cname', + value: element.cname + }); + md.ssrcs.push({ + id: Number(element.ssrc), + attribute: 'label', + value: element.label + }); + md.ssrcs.push({ + id: Number(element.ssrc), + attribute: 'mslabel', + value: element.mslabel + }); + md.ssrcs.push({ + id: Number(element.ssrc), + attribute: 'msid', + value: `${element.mslabel} ${element.label}` + }); + offer.media.push(md); + } + } + protected addIngestMids(offer: SessionDescription) { if (!this.endpointDescription) { throw new Error('Missing endpointDescription'); @@ -236,6 +366,8 @@ export class Connection { offer.media.push(audioDescription); } + + this.addVideoMid(offer); } protected addSFUMids(offer: SessionDescription) { diff --git a/src/db/couchdb.test.ts b/src/db/couchdb.test.ts index d2eb129..ba2edd6 100644 --- a/src/db/couchdb.test.ts +++ b/src/db/couchdb.test.ts @@ -457,6 +457,130 @@ describe('DbManagerCouchDb.saveUserSession', () => { }); }); +/** + * The `session_` prefix belongs to the document id and must never reach the + * session id callers hold. When it did, the id returned on join (bare) and the + * id listed for the same participant (prefixed) compared unequal, so a client + * excluding itself by id never matched and could act on its own session. + */ +describe('DbManagerCouchDb session id / document id separation', () => { + const RAW_ID = '38ef5a00-4491-4b4e-9f2a-1c0d9d2b7a10'; + const DOC_ID = `session_${RAW_ID}`; + + it('stores under the prefixed doc id but does not alter the caller id', async () => { + const { manager, nanoDb } = createTestManager(); + const notFound: any = new Error('not_found'); + notFound.statusCode = 404; + nanoDb.get.mockRejectedValueOnce(notFound); + nanoDb.insert.mockResolvedValueOnce({ ok: true }); + + await manager.saveUserSession(RAW_ID, { + name: 'alpha', + productionId: '1', + lineId: '1', + isWhip: false + } as any); + + // Looked up and written under the prefixed document id... + expect(nanoDb.get).toHaveBeenCalledWith(DOC_ID); + expect(nanoDb.insert.mock.calls[0][0]._id).toBe(DOC_ID); + }); + + it('round-trips a raw session id unchanged through getSession', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.get.mockResolvedValueOnce({ + _id: DOC_ID, + _rev: '1-abc', + name: 'alpha', + productionId: '1', + lineId: '1' + }); + + const session = await manager.getSession(RAW_ID); + + expect(nanoDb.get).toHaveBeenCalledWith(DOC_ID); + // ...and read back as the bare id the caller passed in. + expect(session?._id).toBe(RAW_ID); + expect(session?._id).not.toContain('session_'); + }); + + it('strips the prefix for every session from getSessionsByQuery', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.find.mockResolvedValueOnce({ + docs: [ + { _id: DOC_ID, name: 'alpha', pinnedVideoSessionId: undefined }, + { _id: 'session_beta-uuid', name: 'beta' } + ] + }); + + const sessions = await manager.getSessionsByQuery({ lineId: '1' } as any); + + expect(sessions.map((s) => s._id)).toEqual([RAW_ID, 'beta-uuid']); + }); + + it('self-exclusion by id works across join and list, the bug this prevents', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.find.mockResolvedValueOnce({ + docs: [{ _id: DOC_ID, name: 'alpha' }] + }); + + // RAW_ID is what the join API handed this client. + const sessions = await manager.getSessionsByQuery({ lineId: '1' } as any); + const others = sessions.filter((s) => s._id !== RAW_ID); + + expect(others).toHaveLength(0); + }); + + it('tolerates an already-prefixed id without double-prefixing', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.get.mockResolvedValueOnce({ _id: DOC_ID, _rev: '1-abc' }); + + const session = await manager.getSession(DOC_ID); + + expect(nanoDb.get).toHaveBeenCalledWith(DOC_ID); + expect(session?._id).toBe(RAW_ID); + }); + + it('deletes by the prefixed doc id when given a raw id', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.get.mockResolvedValueOnce({ _id: DOC_ID, _rev: '2-def' }); + nanoDb.destroy.mockResolvedValueOnce({ ok: true }); + + await manager.deleteUserSession(RAW_ID); + + expect(nanoDb.get).toHaveBeenCalledWith(DOC_ID); + expect(nanoDb.destroy).toHaveBeenCalledWith(DOC_ID, '2-def'); + }); + + it('updates by the prefixed doc id and keeps it on the written doc', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.get.mockResolvedValueOnce({ _id: DOC_ID, _rev: '3-ghi', name: 'a' }); + nanoDb.insert.mockResolvedValueOnce({ ok: true }); + + await manager.updateSession(RAW_ID, { isActive: false } as any); + + expect(nanoDb.get).toHaveBeenCalledWith(DOC_ID); + expect(nanoDb.insert.mock.calls[0][0]._id).toBe(DOC_ID); + }); + + it('does not let a normalized session overwrite the document id on re-save', async () => { + const { manager, nanoDb } = createTestManager(); + nanoDb.get.mockResolvedValueOnce({ _id: DOC_ID, _rev: '1-abc' }); + nanoDb.insert.mockResolvedValueOnce({ ok: true }); + + // A session previously read back (so its _id is bare) written out again. + await manager.saveUserSession(RAW_ID, { + _id: RAW_ID, + name: 'alpha', + productionId: '1', + lineId: '1', + isWhip: false + } as any); + + expect(nanoDb.insert.mock.calls[0][0]._id).toBe(DOC_ID); + }); +}); + describe('DbManagerCouchDb.connect', () => { beforeEach(() => { jest.useFakeTimers(); diff --git a/src/db/couchdb.ts b/src/db/couchdb.ts index 0ade56d..3c03dc3 100644 --- a/src/db/couchdb.ts +++ b/src/db/couchdb.ts @@ -13,6 +13,24 @@ import nano from 'nano'; import { v4 as uuidv4 } from 'uuid'; const SESSION_PRUNE_SECONDS = 7_200; + +const SESSION_DOC_PREFIX = 'session_'; + +const toSessionDocId = (sessionId: string): string => + sessionId.startsWith(SESSION_DOC_PREFIX) + ? sessionId + : `${SESSION_DOC_PREFIX}${sessionId}`; + +const toUserSession = (doc: unknown): UserSession => { + const raw = doc as { _id?: unknown }; + const docId = String(raw?._id ?? ''); + return { + ...(doc as object), + _id: docId.startsWith(SESSION_DOC_PREFIX) + ? docId.slice(SESSION_DOC_PREFIX.length) + : docId + } as UserSession; +}; export class DbManagerCouchDb implements DbManager { private client; private nanoDb: nano.DocumentScope | undefined; @@ -463,28 +481,28 @@ export class DbManagerCouchDb implements DbManager { throw new Error('Database not connected'); } - if (!sessionId.startsWith('session')) { - sessionId = `session_${sessionId}`; - } + const sessionDocId = toSessionDocId(sessionId); let existingDoc: any; // Check if document exists, if not creates new session try { - existingDoc = await this.withRetry(() => this.nanoDb!.get(sessionId)); + existingDoc = await this.withRetry(() => this.nanoDb!.get(sessionDocId)); } catch (error: any) { if (error.statusCode === 404) { - existingDoc = { _id: sessionId }; + existingDoc = { _id: sessionDocId }; } else { throw error; } } const now = new Date(); + // `_id` last on purpose: userSession may be a normalized session carrying a + // bare `_id`, which must not become the document id. const updatedSession = { ...existingDoc, ...userSession, lastSeenAt: now.toISOString(), - _id: sessionId + _id: sessionDocId }; // Set createdAt only on first insert (like MongoDB's $setOnInsert) if (!existingDoc.createdAt) { @@ -498,10 +516,8 @@ export class DbManagerCouchDb implements DbManager { if (!this.nanoDb) { throw new Error('Database not connected'); } - if (!sessionId.startsWith('session')) { - sessionId = `session_${sessionId}`; - } - const session = await this.withRetry(() => this.nanoDb!.get(sessionId)); + const sessionDocId = toSessionDocId(sessionId); + const session = await this.withRetry(() => this.nanoDb!.get(sessionDocId)); const response = await this.withRetry(() => this.nanoDb!.destroy(session._id, session._rev) ); @@ -514,11 +530,9 @@ export class DbManagerCouchDb implements DbManager { throw new Error('Database not connected'); } - if (!sessionId.startsWith('session')) { - sessionId = `session_${sessionId}`; - } - const session = await this.withRetry(() => this.nanoDb!.get(sessionId)); - return session as any as UserSession; + const sessionDocId = toSessionDocId(sessionId); + const session = await this.withRetry(() => this.nanoDb!.get(sessionDocId)); + return toUserSession(session); } async updateSession( @@ -530,13 +544,11 @@ export class DbManagerCouchDb implements DbManager { if (!this.nanoDb) { throw new Error('Database not connected'); } - if (!sessionId.startsWith('session')) { - sessionId = `session_${sessionId}`; - } + const sessionDocId = toSessionDocId(sessionId); let doc: any; try { - doc = await this.withRetry(() => this.nanoDb!.get(sessionId)); + doc = await this.withRetry(() => this.nanoDb!.get(sessionDocId)); } catch (error: any) { if (error.statusCode === 404) { return false; @@ -572,7 +584,7 @@ export class DbManagerCouchDb implements DbManager { const response = await this.withRetry(() => this.nanoDb!.find({ selector, limit: 10000 }) ); - return response.docs as unknown as UserSession[]; // could also expand type UserSession to avoid unknown + return response.docs.map(toUserSession); } async addPreset(preset: Omit): Promise { diff --git a/src/media_streams_info.ts b/src/media_streams_info.ts index e1d213d..0a55796 100644 --- a/src/media_streams_info.ts +++ b/src/media_streams_info.ts @@ -14,6 +14,9 @@ export interface MediaStreamsInfo { audio: { ssrcs: MediaStreamsInfoSsrc[]; }; + video?: { + ssrcs: MediaStreamsInfoSsrc[]; + }; } export type RtpCodec = { diff --git a/src/mock-smb-protocol.ts b/src/mock-smb-protocol.ts index 75cbc61..80ba0a3 100644 --- a/src/mock-smb-protocol.ts +++ b/src/mock-smb-protocol.ts @@ -27,11 +27,13 @@ export class MockSmbProtocol implements ISmbProtocol { conferenceId: string, endpointId: string, audio: boolean, - _data: boolean, + video: boolean, + data: boolean, _iceControlling: boolean, _relayType: 'ssrc-rewrite' | 'forwarder' | 'mixed', _idleTimeout: number, - _smbKey: string + _smbKey: string, + _videoRelayType?: 'ssrc-rewrite' | 'forwarder' | 'mixed' ): Promise { const endpoint: SmbEndpoint = { 'bundle-transport': { @@ -80,18 +82,22 @@ export class MockSmbProtocol implements ISmbProtocol { } ] }, - video: { - ssrcs: [], - 'payload-type': { - id: 100, - name: 'VP8', - clockrate: 90000, - parameters: {}, - 'rtcp-fbs': [{ type: 'nack', subtype: '' }] - }, - 'rtp-hdrexts': [] - }, - data: _data ? { port: 5000 } : undefined + // Mirror real SMB: omit the video block entirely when video=false so + // tests for no-video paths don't get a false-positive video allocation. + ...(video && { + video: { + ssrcs: [], + 'payload-type': { + id: 100, + name: 'VP8', + clockrate: 90000, + parameters: {}, + 'rtcp-fbs': [{ type: 'nack', subtype: '' }] + }, + 'rtp-hdrexts': [] + } + }), + data: data ? { port: 5000 } : undefined }; const conf = this.conferences.get(conferenceId); @@ -181,6 +187,37 @@ export class MockSmbProtocol implements ISmbProtocol { conf.set(endpointId, endpointDescription); } + async reconfigureEndpoint( + _smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpoint, + _smbKey: string + ): Promise { + const conf = this.conferences.get(conferenceId); + if (!conf) { + throw new Error( + `Conference ${conferenceId} not found in MockSmbProtocol` + ); + } + conf.set(endpointId, endpointDescription); + } + + async requestKeyframe( + _smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpoint, + _smbKey: string + ): Promise { + // Mirror the real client: leave the endpoint in its target (re-applied) + // configuration after the remove -> re-add cycle. + const conf = this.conferences.get(conferenceId); + if (conf) { + conf.set(endpointId, endpointDescription); + } + } + async getConferences(_smbUrl: string, _smbKey: string): Promise { return Array.from(this.conferences.keys()); } diff --git a/src/models.ts b/src/models.ts index b38f1ea..e152c78 100644 --- a/src/models.ts +++ b/src/models.ts @@ -83,14 +83,16 @@ export const NewProduction = Type.Object({ lines: Type.Array( Type.Object({ name: Type.String({ minLength: 1, maxLength: 200 }), - programOutputLine: Type.Optional(Type.Boolean()) + programOutputLine: Type.Optional(Type.Boolean()), + videoEnabled: Type.Optional(Type.Boolean()) }) ) }); export const NewProductionLine = Type.Object({ name: Type.String({ minLength: 1, maxLength: 200 }), - programOutputLine: Type.Optional(Type.Boolean()) + programOutputLine: Type.Optional(Type.Boolean()), + videoEnabled: Type.Optional(Type.Boolean()) }); const SmbCandidate = Type.Object({ @@ -151,11 +153,7 @@ const SmbRtpHeaderExtension = Type.Object({ uri: Type.String() }); -const VideoSmbPayloadParameters = Type.Object({ - 'x-google-start-bitrate': Type.Optional(Type.String()), - 'x-google-max-bitrate': Type.Optional(Type.String()), - 'x-google-min-bitrate': Type.Optional(Type.String()) -}); +const VideoSmbPayloadParameters = Type.Record(Type.String(), Type.String()); const VideoSmbPayloadType = Type.Object({ id: Type.Number(), @@ -189,11 +187,30 @@ export const SmbEndpointDescription = Type.Object({ 'payload-type': AudioSmbPayloadType, 'rtp-hdrexts': Type.Array(SmbRtpHeaderExtension) }), - video: Type.Object({ - ssrcs: Type.Array(Type.Number()), - 'payload-type': VideoSmbPayloadType, - 'rtp-hdrexts': Type.Array(SmbRtpHeaderExtension) - }), + + video: Type.Optional( + Type.Object({ + ssrcs: Type.Optional(Type.Array(Type.Number())), + 'payload-type': Type.Optional(VideoSmbPayloadType), + 'payload-types': Type.Optional(Type.Array(VideoSmbPayloadType)), + 'rtp-hdrexts': Type.Optional(Type.Array(SmbRtpHeaderExtension)), + streams: Type.Optional( + Type.Array( + Type.Object({ + id: Type.String(), + content: Type.String(), + sources: Type.Array( + Type.Object({ + main: Type.Number(), + feedback: Type.Optional(Type.Number()) + }) + ) + }) + ) + ), + 'ssrc-whitelist': Type.Optional(Type.Array(Type.Number())) + }) + ), data: Type.Optional(Type.Object({ port: Type.Number() })), idleTimeout: Type.Optional(Type.Number()) }); @@ -227,7 +244,9 @@ export const UserResponse = Type.Object({ sessionId: Type.String(), endpointId: Type.Optional(Type.String()), isActive: Type.Boolean(), - isWhip: Type.Boolean() + isWhip: Type.Boolean(), + isWhepReceiver: Type.Optional(Type.Boolean()), + hasVideo: Type.Boolean() }); export const UserSession = Type.Object({ @@ -249,7 +268,10 @@ export const UserSession = Type.Object({ endpointId: Type.Optional(Type.String()), sessionDescription: Type.Optional(SmbEndpointDescription), iceCandidates: Type.Optional(Type.Array(IceCandidate)), - isWhip: Type.Boolean() + isWhip: Type.Boolean(), + isWhepReceiver: Type.Optional(Type.Boolean()), + hasVideo: Type.Boolean(), + pinnedVideoSessionId: Type.Optional(Type.String()) }); export const Conference = Type.Object({ @@ -262,7 +284,9 @@ export const Line = Type.Object({ name: Type.String(), id: Type.String(), smbConferenceId: Type.String(), - programOutputLine: Type.Optional(Type.Boolean()) + programOutputLine: Type.Optional(Type.Boolean()), + videoEnabled: Type.Optional(Type.Boolean()), + whepSourceSessionId: Type.Optional(Type.Union([Type.String(), Type.Null()])) }); export const LineResponse = Type.Object({ @@ -270,12 +294,32 @@ export const LineResponse = Type.Object({ id: Type.String(), smbConferenceId: Type.String(), participants: Type.Array(UserResponse), - programOutputLine: Type.Optional(Type.Boolean()) + programOutputLine: Type.Optional(Type.Boolean()), + videoEnabled: Type.Optional(Type.Boolean()), + whepSourceSessionId: Type.Optional(Type.Union([Type.String(), Type.Null()])) }); export const PatchLine = Type.Omit(Line, ['id', 'smbConferenceId']); export const PatchLineResponse = Type.Omit(Line, ['smbConferenceId']); +export const SetLineWhepSourceRequest = Type.Object({ + pinnedSessionId: Type.Union([Type.String(), Type.Null()]) +}); + +export const SetLineWhepSourceResponse = Type.Object({ + lineId: Type.String(), + pinnedSessionId: Type.Union([Type.String(), Type.Null()]) +}); + +export const SetSessionVideoSourceRequest = Type.Object({ + pinnedSessionId: Type.Union([Type.String(), Type.Null()]) +}); + +export const SetSessionVideoSourceResponse = Type.Object({ + sessionId: Type.String(), + pinnedSessionId: Type.Union([Type.String(), Type.Null()]) +}); + export const Production = Type.Object({ _id: Type.Number(), name: Type.String(), diff --git a/src/production_manager.test.ts b/src/production_manager.test.ts index 78a5c07..5f56d1a 100644 --- a/src/production_manager.test.ts +++ b/src/production_manager.test.ts @@ -348,6 +348,7 @@ describe('production_manager', () => { { name: 'newName', programOutputLine: false, + videoEnabled: false, id: '2', smbConferenceId: '' } diff --git a/src/production_manager.ts b/src/production_manager.ts index 2beb2a9..c3a610f 100644 --- a/src/production_manager.ts +++ b/src/production_manager.ts @@ -257,7 +257,8 @@ export class ProductionManager extends EventEmitter { name: line.name, id: index.toString(), smbConferenceId: '', - programOutputLine: line.programOutputLine || false + programOutputLine: line.programOutputLine || false, + videoEnabled: line.videoEnabled || false }; newProductionLines.push(newProductionLine); } @@ -276,7 +277,8 @@ export class ProductionManager extends EventEmitter { async addProductionLine( production: Production, newLineName: string, - programOutputLine: boolean + programOutputLine: boolean, + videoEnabled = false ): Promise { const nextLineId = production.lines.length ? Math.max(...production.lines.map((line) => parseInt(line.id, 10))) + 1 @@ -286,7 +288,8 @@ export class ProductionManager extends EventEmitter { name: newLineName, id: nextLineId.toString(), smbConferenceId: '', - programOutputLine: programOutputLine || false + programOutputLine: programOutputLine || false, + videoEnabled: videoEnabled || false }); return this.dbManager.updateProduction(production); @@ -305,6 +308,57 @@ export class ProductionManager extends EventEmitter { return undefined; } + async clearWhepSourceIfPinned(sessionId: string): Promise { + try { + const session = await this.dbManager.getSession(sessionId); + if (!session) return; + const productionIdNum = parseInt(session.productionId, 10); + if (Number.isNaN(productionIdNum)) return; + const production = await this.getProduction(productionIdNum); + if (!production) return; + const line = production.lines.find((l) => l.id === session.lineId); + if (!line) return; + if (line.whepSourceSessionId !== sessionId) return; + line.whepSourceSessionId = null; + await this.dbManager.updateProduction(production); + } catch (err) { + Log().warn( + `[whep-pin] clearWhepSourceIfPinned for ${sessionId} failed: ${err}` + ); + } + } + + async getReceiversPinnedToSession( + leaverSessionId: string + ): Promise { + const leaver = await this.dbManager.getSession(leaverSessionId); + if (!leaver) return []; + const sessions = await this.dbManager.getSessionsByQuery({ + productionId: leaver.productionId, + lineId: leaver.lineId, + isExpired: false + }); + return sessions.filter( + (s) => + (s as any)._id?.toString?.() !== leaverSessionId && + (s as any).pinnedVideoSessionId === leaverSessionId + ); + } + + async setLineWhepSource( + production: Production, + lineId: string, + sessionId: string | null + ): Promise { + const line = production.lines.find((l) => l.id === lineId); + if (!line) return undefined; + const current = line.whepSourceSessionId ?? null; + const next = sessionId ?? null; + if (current === next) return production; + line.whepSourceSessionId = sessionId; + return this.dbManager.updateProduction(production); + } + async deleteProductionLine( production: Production, lineId: string @@ -379,7 +433,11 @@ export class ProductionManager extends EventEmitter { lineId: string, sessionId: string, name: string, - isWhip = false + isWhip = false, + // WHEP recipients also currently set isWhip=true (legacy from when both + // WHIP and WHEP shared a code path). + isWhepReceiver = false, + hasVideo = false ): Promise { const userSession: UserSession = { _id: sessionId, @@ -390,7 +448,9 @@ export class ProductionManager extends EventEmitter { lastSeen: isWhip ? Date.now() + 20000 : Date.now(), isActive: true, isExpired: false, - isWhip + isWhip, + isWhepReceiver, + hasVideo }; this.userSessions[sessionId] = userSession; @@ -440,27 +500,63 @@ export class ProductionManager extends EventEmitter { endpointId: string, sessionDescription: SmbEndpointDescription ): Promise { + const smbPresenceKey = endpointId.toLowerCase(); + + const ok = await this.dbManager.updateSession(sessionId, { + endpointId, + sessionDescription, + isActive: true, + isExpired: false, + lastSeen: Date.now(), + ...({ smbPresenceKey } as any) + }); + const userSession = this.userSessions[sessionId]; if (userSession) { userSession.endpointId = endpointId; userSession.sessionDescription = sessionDescription; - const smbPresenceKey = endpointId.toLowerCase(); - (userSession as any).smbPresenceKey = smbPresenceKey; + } - const ok = await this.dbManager.updateSession(sessionId, { - endpointId, - sessionDescription, - isActive: true, - isExpired: false, - lastSeen: Date.now(), - ...({ smbPresenceKey } as any) - }); + if (ok) this.emit('users:change'); + return ok; + } + + async updateSessionHasVideo( + sessionId: string, + hasVideo: boolean + ): Promise { + const ok = await this.dbManager.updateSession(sessionId, { + hasVideo + } as any); + + const userSession = this.userSessions[sessionId]; + if (userSession) { + userSession.hasVideo = hasVideo; + } + + if (ok) this.emit('users:change'); + return ok; + } - if (ok) this.emit('users:change'); - return ok; + async updateSessionVideoPin( + sessionId: string, + sessionDescription: SmbEndpointDescription, + pinnedVideoSessionId: string | null + ): Promise { + const ok = await this.dbManager.updateSession(sessionId, { + sessionDescription, + ...({ pinnedVideoSessionId } as any) + }); + + const userSession = this.userSessions[sessionId]; + if (userSession) { + userSession.sessionDescription = sessionDescription; + (userSession as any).pinnedVideoSessionId = pinnedVideoSessionId; } - return false; + + if (ok) this.emit('users:change'); + return ok; } removeUserSession(sessionId: string): string | undefined { @@ -491,7 +587,9 @@ export class ProductionManager extends EventEmitter { sessionId: s._id?.toString?.() ?? '', name: s.name ?? '', isActive: !!s.isActive, - isWhip: !!s.isWhip + isWhip: !!s.isWhip, + isWhepReceiver: !!s.isWhepReceiver, + hasVideo: !!s.hasVideo }; if (typeof s.endpointId === 'string' && s.endpointId.length > 0) u.endpointId = s.endpointId; @@ -511,4 +609,11 @@ export class ProductionManager extends EventEmitter { return participants; } + + async getUserNameBySessionId(sessionId: string): Promise { + const cached = this.userSessions[sessionId]; + if (cached) return cached.name; + const dbSession = await this.dbManager.getSession(sessionId); + return dbSession?.name ?? null; + } } diff --git a/src/sfu/constants.ts b/src/sfu/constants.ts new file mode 100644 index 0000000..fb3d8ef --- /dev/null +++ b/src/sfu/constants.ts @@ -0,0 +1,2 @@ +export const NORMALIZED_VIDEO_PT_MAIN = 96; +export const NORMALIZED_VIDEO_PT_RTX = 97; diff --git a/src/sfu/interface.ts b/src/sfu/interface.ts index 2e1d160..f7ab1b3 100644 --- a/src/sfu/interface.ts +++ b/src/sfu/interface.ts @@ -66,6 +66,14 @@ export interface SfuVideoStream { content: string; } +export interface VideoSmbPayloadType { + id: number; + name: string; + clockrate: number; + parameters: Record; + 'rtcp-fbs': { type: string; subtype?: string }[]; +} + export interface SfuEndpointDescription { 'bundle-transport'?: SfuTransport; audio: { @@ -73,7 +81,14 @@ export interface SfuEndpointDescription { 'payload-type': AudioSmbPayloadType; 'rtp-hdrexts': SfuRtpHeaderExtension[]; }; - + video?: { + 'payload-type'?: VideoSmbPayloadType; + 'payload-types'?: VideoSmbPayloadType[]; + 'rtp-hdrexts'?: SfuRtpHeaderExtension[]; + ssrcs?: number[]; + streams?: SfuVideoStream[]; + 'ssrc-whitelist'?: number[]; + }; data?: { port: number; }; diff --git a/src/smb.test.ts b/src/smb.test.ts index 758f382..a6e0d08 100644 --- a/src/smb.test.ts +++ b/src/smb.test.ts @@ -1,4 +1,4 @@ -import { SmbProtocol } from './smb'; +import { SmbEndpointActionError, SmbProtocol } from './smb'; // Mock fetch globally const mockFetch = jest.fn(); @@ -113,6 +113,7 @@ describe('SmbProtocol', () => { 'conf-1', 'ep-1', true, + false, true, true, 'ssrc-rewrite', @@ -142,6 +143,7 @@ describe('SmbProtocol', () => { 'ep-1', true, false, + false, true, 'mixed', 60, @@ -160,6 +162,7 @@ describe('SmbProtocol', () => { 'conf-1', 'ep-1', false, + false, true, true, 'ssrc-rewrite', @@ -180,6 +183,7 @@ describe('SmbProtocol', () => { 'ep-1', true, false, + false, true, 'ssrc-rewrite', 120, @@ -199,6 +203,7 @@ describe('SmbProtocol', () => { 'conf-1', 'ep-1', true, + false, true, true, 'ssrc-rewrite', @@ -334,6 +339,138 @@ describe('SmbProtocol', () => { ) ).rejects.toThrow('Failed to configure endpoint'); }); + + // The route turns this specific rejection into a 425 so the client retries. + // It matches on the typed error, so the client must actually produce one — + // a plain Error here would silently turn the retry back into a 500. + it('rejects with a typed error that flags a not-yet-configured endpoint', async () => { + mockFetch.mockResolvedValue( + // text() is what the client reads, so the body must be the raw JSON + // string SMB actually sends. + mockResponse( + 400, + JSON.stringify({ + message: + "Can't reconfigure audio because it was not configured in first place", + status_code: 400 + }), + 'application/json' + ) + ); + + const err = await smb + .reconfigureEndpoint( + smbUrl, + 'conf-1', + 'ep-1', + endpointDescription as any, + smbKey + ) + .catch((e) => e); + + expect(err).toBeInstanceOf(SmbEndpointActionError); + expect(err.status).toBe(400); + expect(err.action).toBe('reconfigure'); + expect(err.isEndpointNotConfiguredYet).toBe(true); + }); + + it('does not flag an unrelated 400 as not-yet-configured', async () => { + mockFetch.mockResolvedValue( + mockResponse( + 400, + JSON.stringify({ message: 'invalid ssrc' }), + 'application/json' + ) + ); + + const err = await smb + .reconfigureEndpoint( + smbUrl, + 'conf-1', + 'ep-1', + endpointDescription as any, + smbKey + ) + .catch((e) => e); + + expect(err).toBeInstanceOf(SmbEndpointActionError); + expect(err.isEndpointNotConfiguredYet).toBe(false); + }); + }); + + // ── requestKeyframe ──────────────────────────────────────────────── + + describe('requestKeyframe', () => { + const pinnedDescription = { + audio: { ssrcs: [9000] }, + video: { ssrcs: [], 'ssrc-whitelist': [3333, 4444] } + }; + + it('performs a whitelist remove -> re-add reconfigure cycle', async () => { + mockFetch.mockResolvedValue(mockResponse(200, null)); + + await smb.requestKeyframe( + smbUrl, + 'conf-1', + 'ep-1', + pinnedDescription as any, + smbKey + ); + + // Two reconfigure PUTs: clear, then re-apply. + expect(mockFetch).toHaveBeenCalledTimes(2); + + const first = JSON.parse(mockFetch.mock.calls[0][1].body); + expect(first.action).toBe('reconfigure'); + expect(first.video['ssrc-whitelist']).toBeUndefined(); + + const second = JSON.parse(mockFetch.mock.calls[1][1].body); + expect(second.action).toBe('reconfigure'); + expect(second.video['ssrc-whitelist']).toEqual([3333, 4444]); + }); + + it('does not mutate the original endpointDescription', async () => { + mockFetch.mockResolvedValue(mockResponse(200, null)); + const original = JSON.parse(JSON.stringify(pinnedDescription)); + + await smb.requestKeyframe( + smbUrl, + 'conf-1', + 'ep-1', + pinnedDescription as any, + smbKey + ); + + expect(pinnedDescription).toEqual(original); + }); + + it('is a no-op when there is no ssrc-whitelist', async () => { + mockFetch.mockResolvedValue(mockResponse(200, null)); + + await smb.requestKeyframe( + smbUrl, + 'conf-1', + 'ep-1', + { audio: { ssrcs: [9000] }, video: { ssrcs: [] } } as any, + smbKey + ); + + expect(mockFetch).not.toHaveBeenCalled(); + }); + + it('is a no-op when there is no video block', async () => { + mockFetch.mockResolvedValue(mockResponse(200, null)); + + await smb.requestKeyframe( + smbUrl, + 'conf-1', + 'ep-1', + { audio: { ssrcs: [9000] } } as any, + smbKey + ); + + expect(mockFetch).not.toHaveBeenCalled(); + }); }); // ── getConferences ───────────────────────────────────────────────── diff --git a/src/smb.ts b/src/smb.ts index 2cb52ef..eb2a8f8 100644 --- a/src/smb.ts +++ b/src/smb.ts @@ -10,6 +10,23 @@ interface AllocateConferenceResponse { id: string; } +export class SmbEndpointActionError extends Error { + constructor( + readonly action: 'configure' | 'reconfigure', + readonly status: number, + readonly body: string + ) { + super(`Failed to ${action} endpoint: status=${status} body=${body}`); + this.name = 'SmbEndpointActionError'; + } + + get isEndpointNotConfiguredYet(): boolean { + return ( + this.status === 400 && /not configured in first place/i.test(this.body) + ); + } +} + interface BaseAllocationRequest { action: string; 'bundle-transport': { @@ -32,17 +49,23 @@ interface AudioAllocationRequest { } export interface ISmbProtocol { - allocateConference(smbUrl: string, smbKey: string): Promise; + allocateConference( + smbUrl: string, + smbKey: string, + lastN?: number + ): Promise; allocateEndpoint( smbUrl: string, conferenceId: string, endpointId: string, audio: boolean, + video: boolean, data: boolean, iceControlling: boolean, relayType: 'ssrc-rewrite' | 'forwarder' | 'mixed', idleTimeout: number, - smbKey: string + smbKey: string, + videoRelayType?: 'ssrc-rewrite' | 'forwarder' | 'mixed' ): Promise; allocateAudioEndpoint( smbUrl: string, @@ -59,6 +82,20 @@ export interface ISmbProtocol { endpointDescription: SmbEndpointDescription, smbKey: string ): Promise; + reconfigureEndpoint( + smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpointDescription, + smbKey: string + ): Promise; + requestKeyframe( + smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpointDescription, + smbKey: string + ): Promise; getConferences(smbUrl: string, smbKey: string): Promise; getConferencesWithUsers( smbUrl: string, @@ -72,14 +109,22 @@ export interface ISmbProtocol { } export class SmbProtocol implements ISmbProtocol { - async allocateConference(smbUrl: string, smbKey: string): Promise { + async allocateConference( + smbUrl: string, + smbKey: string, + lastN?: number + ): Promise { + const requestBody: Record = {}; + if (typeof lastN === 'number' && lastN > 0) { + requestBody['last-n'] = lastN; + } const allocateResponse = await fetch(smbUrl, { method: 'POST', headers: { 'Content-Type': 'application/json', ...(smbKey !== '' && { Authorization: `Bearer ${smbKey}` }) }, - body: '{}' + body: JSON.stringify(requestBody) }); if (!allocateResponse.ok) { @@ -100,11 +145,13 @@ export class SmbProtocol implements ISmbProtocol { conferenceId: string, endpointId: string, audio: boolean, + video: boolean, data: boolean, iceControlling: boolean, relayType: 'ssrc-rewrite' | 'forwarder' | 'mixed', idleTimeout: number, - smbKey: string + smbKey: string, + videoRelayType?: 'ssrc-rewrite' | 'forwarder' | 'mixed' ): Promise { const request: BaseAllocationRequest = { action: 'allocate', @@ -113,12 +160,6 @@ export class SmbProtocol implements ISmbProtocol { ice: true, dtls: true, sdes: false - }, - audio: { - ssrcs: [] - }, - video: { - ssrcs: [] } }; @@ -126,6 +167,10 @@ export class SmbProtocol implements ISmbProtocol { request['audio'] = { 'relay-type': relayType }; } + if (video) { + request['video'] = { 'relay-type': videoRelayType ?? relayType }; + } + if (data) { request['data'] = {}; } @@ -207,7 +252,8 @@ export class SmbProtocol implements ISmbProtocol { return smbEndpointDescription; } - async configureEndpoint( + private async sendEndpointAction( + action: 'configure' | 'reconfigure', smbUrl: string, conferenceId: string, endpointId: string, @@ -215,8 +261,9 @@ export class SmbProtocol implements ISmbProtocol { smbKey: string ): Promise { const request = JSON.parse(JSON.stringify(endpointDescription)); - request['action'] = 'configure'; + request['action'] = action; const url = smbUrl + conferenceId + '/' + endpointId; + const response = await fetch(url, { method: 'PUT', headers: { @@ -227,21 +274,84 @@ export class SmbProtocol implements ISmbProtocol { }); if (!response.ok) { - const contentType = response.headers.get('content-type'); + const body = await response.text(); + throw new SmbEndpointActionError(action, response.status, body); + } + } - let text; - let json; + async configureEndpoint( + smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpointDescription, + smbKey: string + ): Promise { + return this.sendEndpointAction( + 'configure', + smbUrl, + conferenceId, + endpointId, + endpointDescription, + smbKey + ); + } - if (contentType && contentType.indexOf('text/plain') > -1) { - text = await response.text(); - } else if (contentType && contentType.indexOf('application/json') > -1) { - json = await response.json(); - } + async reconfigureEndpoint( + smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpointDescription, + smbKey: string + ): Promise { + return this.sendEndpointAction( + 'reconfigure', + smbUrl, + conferenceId, + endpointId, + endpointDescription, + smbKey + ); + } - throw new Error( - `Failed to configure endpoint ${text ? text : JSON.stringify(json)}` - ); + async requestKeyframe( + smbUrl: string, + conferenceId: string, + endpointId: string, + endpointDescription: SmbEndpointDescription, + smbKey: string + ): Promise { + const targetWhitelist = endpointDescription.video?.['ssrc-whitelist']; + // Nothing to refresh if there is no video block or no pinned source. + if (!endpointDescription.video || !targetWhitelist) { + return; + } + + // Step 1: clear the whitelist so SMB drops the current forwarding context. + const cleared: SmbEndpointDescription = JSON.parse( + JSON.stringify(endpointDescription) + ); + if (cleared.video) { + delete cleared.video['ssrc-whitelist']; } + await this.sendEndpointAction( + 'reconfigure', + smbUrl, + conferenceId, + endpointId, + cleared, + smbKey + ); + + // Step 2: re-apply the target whitelist. The freshly initialized + // forwarding context triggers a PLI to the publisher -> fresh keyframe. + await this.sendEndpointAction( + 'reconfigure', + smbUrl, + conferenceId, + endpointId, + endpointDescription, + smbKey + ); } async getConferences(smbUrl: string, smbKey: string): Promise { diff --git a/src/test-fixtures/sdp-fixtures.ts b/src/test-fixtures/sdp-fixtures.ts index 1a28ade..ef52894 100644 --- a/src/test-fixtures/sdp-fixtures.ts +++ b/src/test-fixtures/sdp-fixtures.ts @@ -62,7 +62,7 @@ const AUDIO_VIDEO_SDP = [ 'a=extmap:1 urn:ietf:params:rtp-hdrext:ssrc-audio-level', 'a=extmap:3 http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time', 'a=candidate:1 1 UDP 2130706431 192.168.1.100 50000 typ host', - 'm=video 9 UDP/TLS/RTP/SAVPF 96 97 98', + 'm=video 9 UDP/TLS/RTP/SAVPF 96 97 98 99', 'c=IN IP4 0.0.0.0', 'a=rtcp:9 IN IP4 0.0.0.0', 'a=ice-ufrag:videoUfrag', @@ -75,8 +75,10 @@ const AUDIO_VIDEO_SDP = [ 'a=rtpmap:96 VP8/90000', 'a=rtpmap:97 rtx/90000', 'a=rtpmap:98 H264/90000', + 'a=rtpmap:99 rtx/90000', 'a=fmtp:97 apt=96', 'a=fmtp:98 profile-level-id=42e01f;level-asymmetry-allowed=1;packetization-mode=1', + 'a=fmtp:99 apt=98', 'a=rtcp-fb:96 nack', 'a=rtcp-fb:96 goog-remb', 'a=rtcp-fb:96 transport-cc', diff --git a/tsconfig.base.json b/tsconfig.base.json index bbc7478..a7c1f1f 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -1,6 +1,7 @@ { "compilerOptions": { - /* Visit https://aka.ms/tsconfig to read more about this file + /* Visit https://aka.ms/tsconfig to read more about this file */ + /* JavaScript Support */ "allowJs": true /* Allow JavaScript files to be a part of your program. Use the 'checkJS' option to get errors from these files. */, "checkJs": true /* Enable error reporting in type-checked JavaScript files. */,