feat: add cross-pod co-broadcaster transport creation

backend/src/handlers/registerBroadcaster.handler.ts

@@ -223,28 +223,120 @@ const registerBroadcasterHandler = async (socket: Socket) => {
     try {
       const room = getRoom(roomId);
       const hostUserId = socket.data.user?.id; 
-      const router = room.router;
+      const socketId = socket.id; 
 
       const roomKey = `room:${roomId}`; 
-      const redisRoom = await getRedisRoom(roomKey)
-      //TODO: Implementation for different pod connections
-      if(redisRoom.nodeId !== config.instanceId){
-        logger.error('Different pod')
-        throw new ApiError(409,"Room belongs to another node")
+      let redisRoom; 
+      try {
+        redisRoom = await getRedisRoom(roomKey)
+      } catch (error) {
+        logger.error('Error finding redis room',{
+          error: (error as Error).message,
+          stack: (error as Error).stack
+        })
+
+        ack({success: false, code: "TRANSPORT_CREATION_FAILED"})
+        return; 
       }
-
 
+      if(redisRoom?.nodeId !== config.instanceId){
+        const requestId = crypto.randomUUID()
+        const date = Date.now(); 
+        const args = {roomId, socketId}; 
+        const replyTo = `pod:${config.instanceId}:response`; 
+
+        const payLoad : PodCommandPayload = {
+          requestId, 
+          type: 'createBroadcasterTransport', 
+          args, 
+          replyTo, 
+          date
+        }
+
+        const TIMEOUTMS = 5000; 
+        const timeOuthandle = setTimeout(() => {
+          const entry = podRequestHandleMap.get(requestId); 
+          if(!entry) return logger.warn(`Entry not found with requestId: ${requestId}`); 
+          entry.onComplete({}, 'TRANSPORT_CREATION_FAILED'); 
+          podRequestHandleMap.delete(requestId); 
+        }, TIMEOUTMS)
+
+        podRequestHandleMap.set(requestId, {
+          requestId, 
+          socketId, 
+          startDate: date, 
+          requestType: 'createBroadcasterTransport', 
+          status: 'pending',
+          replyTo, 
+
+          onComplete: (result, error) => {
+            clearTimeout(timeOuthandle)
+
+            if(error){
+              logger.error('[Transport] pod error', {
+                error : error
+              })
+
+              ack({success: false, code: 'TRANSPORT_CREATION_FAILED'})
+              return; 
+            }
+
+            ack({success: true, data: result})
+
+            void Broadcaster.findOneAndUpdate(
+              {
+                broadcasterId: hostUserId,
+                roomId,
+              },
+              {
+                $push: {
+                  transportIds: result?.id,
+                },
+              }
+            ).catch((error) => {
+              logger.error("Failed to update broadcaster transport", error);
+            });
+
+          }
+
+        })
+
+        if(!redisRoom.nodeId){
+          logger.error('Redis room nodeId not found'); 
+          podRequestHandleMap.delete(requestId); 
+          ack({success: false, code: 'TRANSPORT_CREATION_FAILED'})
+          return; 
+        }
+
+        let receivers: number;
+        try {
+          receivers = await publishCommand(payLoad, redisRoom.nodeId)
+        } catch (error) {
+          clearTimeout(timeOuthandle);
+          podRequestHandleMap.delete(requestId);
+          throw error;
+        }
+        if(receivers === 0){
+          logger.error('Cross [POD] connection failed'); 
+          clearTimeout(timeOuthandle);
+          podRequestHandleMap.delete(requestId); 
+          ack({success: false, code: 'TRANSPORT_CREATION_FAILED'})
+        }
+        return; 
+      }
+
+      const router = room.router; 
       const broadcasterTransport =
         await createWebRtcTransport(
           router,
           roomId,
-          socket.id,
+          socketId,
           "producer"
         );
 
       await saveBroadcasterTransport(
         roomId,
-        socket.id,
+        socketId,
         broadcasterTransport
       );
 

backend/src/utils/podConnection.ts

@@ -7,7 +7,8 @@
 import { heartBeat } from "./roomCordinator";
 import {consume} from '../handlers/viewer.handler'
 import { pauseConsumer, resumeConsumer } from "../consumer/consumer.handler";
-
+import { createWebRtcTransport } from "../mediasoup/transport";
+import { addBroadcaster, saveBroadcasterTransport } from "./broadcaster.util";
 
 export interface PodCommandPayload {
   type: string;       
@@ -76,7 +77,33 @@
                 break; 
             }
 
-            case(type === ''): {}
+            case(type === 'createBroadcasterTransport'): {
+                const {roomId, socketId} = args as unknown as generalArgs; 
+                const routerId = roomToRouter.get(roomId); 
+                if(!routerId){
+                    error = 'RouterId not found'
+                    throw new Error('RouterId not found')
+                }
+                const router = getRouter(routerId); 
+
+                const broadcasterTransport = await createWebRtcTransport(router,roomId, socketId, 'producer');
+
+                result = {
+                    id: broadcasterTransport?.id,
+                    iceParameters: broadcasterTransport?.iceParameters,
+                    iceCandidates: broadcasterTransport?.iceCandidates,
+                    dtlsParameters: broadcasterTransport?.dtlsParameters,
+                }
+                addBroadcaster(roomId, socketId)
+                await saveBroadcasterTransport(roomId, socketId, broadcasterTransport)
+
+                const payLoad: PodResponsePayload = {
+                    requestId, 
+                    result
+                }
+                await publishResponse(payLoad, replyTo)
+                break; 
+            }
             case(type === ''): {}
             case(type === ''): {}
 
@@ -266,15 +293,16 @@
     podRequestHandleMap.delete(payload.requestId)
 }
 
-export const publishCommand = async(payload: PodCommandPayload, targetNode:string) => {
+export const publishCommand = async(payload: PodCommandPayload, targetNode:string):Promise<number> => {
     const channel = `pod:${targetNode}:cmd`
-    const receivers = await redis.spublish(channel , JSON.stringify(payload))
+    const receivers = await redis.spublish(channel , JSON.stringify(payload)) as number
 
     if(receivers === 0){
         logger.error(`No subscribers for ${channel} — pod may be down`, { requestId: payload.requestId });
     }
 
     logger.info("Redis delivered to", receivers, "subscribers"); 
+
     return receivers; 
 }