server.js

const ws = require('ws');
const MediasoupSignalingDelegate = require('./webrtc/MediasoupSignalingDelegate');
const os = require('os');
const minPort = 5000;
const maxPort = 6000;
const mediaserver = new MediasoupSignalingDelegate();

let users = new Map();
let currentSocket = null;

if (process.argv.length < 3) {
    console.log(`[MEDIA PROXY AGENT] **Error: Listen port required.**

You must specify the port to listen on.

**Proper Usage:**
node server.js <PORT> [use_public_ip_flag]

- **<PORT>**: The port to listen for connections (e.g., 4444).
- **[use_public_ip_flag]**: An optional argument. Pass **true** to use the machine's public IP address for relaying instead of a local one.`);
    return;
}

let listen_port = parseInt(process.argv[2]);
let use_public_ip = process.argv[3] && process.argv[3].toLowerCase() === 'true';
let internal_config;

global.MEDIA_CODECS = [
    {
        kind: 'audio',
        mimeType: 'audio/opus',
        clockRate: 48000,
        channels: 2,
        parameters: {
            'minptime': 10,
            'useinbandfec': 1,
            'usedtx': 1
        },
        preferredPayloadType: 111,
    },
    {
        kind: 'video',
        mimeType: 'video/VP8',
        clockRate: 90000,
        rtcpFeedback: [
            { type: 'ccm', parameter: 'fir' },
            { type: 'nack' },
            { type: 'nack', parameter: 'pli' },
            { type: 'goog-remb' }
        ],
        preferredPayloadType: 101
    }
];

global.onClientJoinedRoom = async (_client) => {
    if (!_client.webrtcConnected || !_client.voiceRoomId || !_client.room || !currentSocket) return;

    let clients = new Set(_client.room._clients.values());
    let video_batch = {};

    await Promise.all(
        Array.from(clients).map(async (client) => {
            if (client.user_id === _client.user_id) return;

            let needsUpdate = false;
            let consumerAudioSsrc = 0;
            let consumerVideoSsrc = 0;
            let consumerRtxSsrc = 0;

            if (client.isProducingAudio() && !_client.isSubscribedToTrack(client.user_id, "audio")) {
                await _client.subscribeToTrack(client.user_id, "audio");
                needsUpdate = true;
            }

            if (client.isProducingVideo() && !_client.isSubscribedToTrack(client.user_id, "video")) {
                await _client.subscribeToTrack(client.user_id, "video");
                needsUpdate = true;
            }

            if (!needsUpdate) return;

            const audioConsumer = _client.consumers.find(
                (consumer) => consumer.producerId === client.audioProducer?.id
            );

            const videoConsumer = _client.consumers.find(
                (consumer) => consumer.producerId === client.videoProducer?.id
            );

            if (audioConsumer) {
                consumerAudioSsrc = audioConsumer.rtpParameters?.encodings?.[0]?.ssrc ?? 0;
            }

            if (videoConsumer) {
                consumerVideoSsrc = videoConsumer.rtpParameters?.encodings?.[0]?.ssrc ?? 0;
                consumerRtxSsrc = videoConsumer.rtpParameters?.encodings?.[0]?.rtx?.ssrc ?? 0;
            }

            video_batch[_client.user_id] = {
                op: 12,
                d: {
                    user_id: client.user_id,
                    audio_ssrc: consumerAudioSsrc,
                    video_ssrc: consumerVideoSsrc,
                    rtx_ssrc: consumerRtxSsrc
                },
            };
        }),
    );

    if (Object.entries(video_batch).length > 0) {
        currentSocket.send(JSON.stringify({
            op: "VIDEO_BATCH",
            d: video_batch
        }));
    }
};

function getIPAddress() {
    var interfaces = os.networkInterfaces();
    for (var devName in interfaces) {
        var iface = interfaces[devName];

        for (var i = 0; i < iface.length; i++) {
            var alias = iface[i];

            if (alias.family === 'IPv4' && alias.address !== '127.0.0.1' && !alias.internal)
                return alias.address;
        }
    }
    return '0.0.0.0';
}

async function start() {
    let ip_address = getIPAddress();
    let lat = 0;
    let lon = 0;

    if (use_public_ip) {
        try {
            let try_get_ip = await fetch("http://ip-api.com/json");
            let data = await try_get_ip.json();
            ip_address = data.query;
            lat = data.lat;
            lon = data.lon;
            console.log(`[MEDIA PROXY AGENT] Detected Public IP: ${ip_address} (Location: ${lat}, ${lon})`);
        } catch (e) {
            console.error(`[MEDIA PROXY AGENT] Failed to fetch public IP/Location: ${e.message}`);
            try {
                let try_get_ip = await fetch("https://checkip.amazonaws.com");
                ip_address = await try_get_ip.text();
                ip_address = ip_address.trim();
            } catch (e2) {
                console.error(`[MEDIA PROXY AGENT] Failed fallback IP fetch: ${e2.message}`);
            }
        }
    } 

    await mediaserver.start(ip_address, minPort, maxPort, true);

    const wss = new ws.WebSocketServer({ port: listen_port });

    console.log(`[MEDIA PROXY AGENT] Listening on port ${listen_port}...`);

    wss.on('connection', (socket) => {
        console.log(`[MEDIA PROXY AGENT] Main server connected!`);
        currentSocket = socket;

        socket.send(JSON.stringify({
            op: "HEARTBEAT_INFO",
            d: {
                heartbeat_interval: 41250
            }
        }));

        socket.send(JSON.stringify({
            op: "IDENTIFY",
            d: {
                public_ip: ip_address,
                public_port: mediaserver.port,
                lat: lat,
                lon: lon,
                timestamp: Date.now()
            }
        }));

        socket.on('message', async (data) => {
            let json = JSON.parse(Buffer.from(data).toString('utf-8'));

            console.log(JSON.stringify(json));

            if (json.op === 'ALRIGHT') {
                let location = json.d.location;

                internal_config = json.d.config;

                console.log(`[MEDIA PROXY AGENT] Identified with main server! There are ${location - 1} other server(s) in front of us.`);
                console.log(`[MEDIA PROXY AGENT] Received configuration from the main server!`);
                console.log(JSON.stringify(internal_config));
            } else if (json.op === 'HEARTBEAT_INFO') {
                let heartbeat_interval = json.d.heartbeat_interval;

                setInterval(() => {
                    socket.send(JSON.stringify({
                        op: "HEARTBEAT",
                        d: Date.now()
                    }));
                }, heartbeat_interval);
            } else if (json.op === 'CLIENT_CLOSE') {
                console.log(`[MEDIA PROXY AGENT] Client closed! Removed from internal store.`);

                users.delete(json.d.user_id);
            } else if (json.op === 'CLIENT_IDENTIFY') {
                let ip_address = json.d.ip_address;
                let user_id = json.d.user_id;
                let ssrc = json.d.ssrc;
                let room_id = json.d.room_id;

                console.log(`[MEDIA PROXY AGENT] Client (${user_id}) joined room id: ${room_id}`);

                let client = await mediaserver.join(room_id, user_id, socket, 'guild-voice');
                
                client.initIncomingSSRCs({
                    audio_ssrc: 0,
                    video_ssrc: 0,
                    rtx_ssrc: 0
                });

                users.set(user_id, {
                    ip_address: ip_address,
                    ssrc: ssrc,
                    room_id: room_id,
                    client: client,
                    is_speaking: false,
                    last_speaking_update: 0
                });
            } else if (json.op === 'OFFER') {
                let sdp = json.d.sdp;
                let codecs = json.d.codecs;
                let ip_address = json.d.ip_address;
                let user_id = json.d.user_id;
                let room_id = json.d.room_id;
                let client_build = json.d.client_build;
                let client_build_date = new Date(json.d.client_build_date);

                let user = users.get(user_id);

                if (!user)  {
                    return;
                }

                let answer = await mediaserver.onOffer(client_build, client_build_date, user.client, sdp, codecs);

                socket.send(JSON.stringify({
                    op: "ANSWER",
                    d: {
                        room_id: room_id,
                        user_id: user_id,
                        sdp: answer.sdp,
                        audio_codec: 'opus',
                        video_codec: answer.selectedVideoCodec
                    }
                }));

                console.log(`[MEDIA PROXY AGENT] Answered client (${user_id})`);
            } else if (json.op === 'CLIENT_SPEAKING') {
                let ip_address = json.d.ip_address;
                let user_id = json.d.user_id;
                let room_id = json.d.room_id;
                let speaking = json.d.speaking;
                let audio_ssrc = json.d.audio_ssrc;

                const now = Date.now();

                let user = users.get(user_id);

                if (!user) {
                    return;
                }

                if (user.is_speaking === speaking && (now - user.last_speaking_update) < internal_config.speaking_throttle_ms) {
                    return;
                }

                user.is_speaking = speaking;

                let speaking_batch = {};
                let video_batch = {};
                let producerClient = user.client;

                if (!producerClient.isProducingAudio()) {
                    if (!json.d.speaking || !audio_ssrc) {
                        return;
                    }

                    console.log(
                        `Client ${user_id} sent a speaking packet but has no audio producer. Attempting to initialize...`,
                    );
                }

                let incomingSSRCs = producerClient.getIncomingStreamSSRCs();

                if (incomingSSRCs.audio_ssrc !== audio_ssrc) {
                    console.log(`[MEDIA PROXY AGENT] [${user_id}] SSRC mismatch detected. Correcting audio SSRC from ${incomingSSRCs.audio_ssrc} to ${audio_ssrc}.`);

                    producerClient.stopPublishingTrack("audio");

                    producerClient.initIncomingSSRCs({
                        audio_ssrc: audio_ssrc,
                        video_ssrc: incomingSSRCs.video_ssrc,
                        rtx_ssrc: incomingSSRCs.rtx_ssrc
                    });

                    await producerClient.publishTrack("audio", { audio_ssrc: audio_ssrc });

                    const clientsToNotify = new Set();

                    for (const otherClient of producerClient.room.clients.values()) {
                        if (otherClient.user_id === user_id) continue;

                        await otherClient.subscribeToTrack(user_id, "audio");

                        clientsToNotify.add(otherClient);
                    }

                    await Promise.all(
                        Array.from(clientsToNotify).map((client) => {
                            const updatedSsrcs = client.getOutgoingStreamSSRCsForUser(user_id);

                            video_batch[client.user_id] = {
                                op: 12,
                                d: {
                                    user_id: user_id,
                                    audio_ssrc: updatedSsrcs.audio_ssrc,
                                    video_ssrc: updatedSsrcs.video_ssrc,
                                    rtx_ssrc: updatedSsrcs.rtx_ssrc
                                }
                            }
                        }),
                    );
                }

                await Promise.all(
                    Array.from(
                        mediaserver.getClientsForRtcServer(
                            room_id,
                        ),
                    ).map((client) => {
                        if (client.user_id === user_id) return Promise.resolve();

                        const ssrcInfo = client.getOutgoingStreamSSRCsForUser(user_id);

                        if (speaking && ssrcInfo.audio_ssrc === 0) {
                            console.log(`[MEDIA PROXY AGENT] Suppressing speaking packet for ${client.user_id} as consumer for ${user_id} is not ready (ssrc=0).`);
                            return Promise.resolve();
                        }

                        speaking_batch[client.user_id] = {
                            op: 5,
                            d: {
                                user_id: user_id,
                                speaking: speaking,
                                ssrc: ssrcInfo.audio_ssrc
                            }
                        }
                    }),
                );

                if (Object.entries(video_batch).length > 0) {
                    socket.send(JSON.stringify({
                        op: "VIDEO_BATCH",
                        d: video_batch
                    }));
                }

                socket.send(JSON.stringify({
                    op: "SPEAKING_BATCH",
                    d: speaking_batch
                }));
            } else if (json.op === 'VIDEO') {
                let user_id = json.d.user_id;
                let d = json.d;

                let user = users.get(user_id);

                if (!user) {
                    return;
                }

                let producerClient = user.client;

                const video_batch = {};
                const clientsThatNeedUpdate = new Set();
                const wantsToProduceAudio = d.audio_ssrc !== 0;
                const wantsToProduceVideo = d.video_ssrc !== 0;

                const isCurrentlyProducingAudio = producerClient.isProducingAudio();
                const isCurrentlyProducingVideo = producerClient.isProducingVideo();

                producerClient.initIncomingSSRCs({
                    audio_ssrc: d.audio_ssrc,
                    video_ssrc: d.video_ssrc,
                    rtx_ssrc: d.rtx_ssrc
                });

                if (wantsToProduceAudio && !isCurrentlyProducingAudio) {
                    console.log(`[MEDIA PROXY AGENT] [${user_id}] Starting audio production with ssrc ${d.audio_ssrc}`);
                    await producerClient.publishTrack("audio", { audio_ssrc: d.audio_ssrc });

                    for (const client of producerClient.room.clients.values()) {
                        if (client.user_id === user_id) continue;
                        await client.subscribeToTrack(user_id, "audio");
                        clientsThatNeedUpdate.add(client);
                    }
                }
                else if (!wantsToProduceAudio && isCurrentlyProducingAudio) {
                    console.log(`[MEDIA PROXY AGENT] [${user_id}] Stopping audio production.`);
                    producerClient.stopPublishingTrack("audio");

                    for (const client of producerClient.room.clients.values()) {
                        if (client.user_id !== user_id) clientsThatNeedUpdate.add(client);
                    }
                }

                if (wantsToProduceVideo && !isCurrentlyProducingVideo) {
                    console.log(`[MEDIA PROXY AGENT] [${user_id}] Starting video production with ssrc ${d.video_ssrc}`);
                    await producerClient.publishTrack("video", { video_ssrc: d.video_ssrc, rtx_ssrc: d.rtx_ssrc });

                    for (const client of producerClient.room.clients.values()) {
                        if (client.user_id === user_id) continue;
                        await client.subscribeToTrack(user_id, "video");
                        clientsThatNeedUpdate.add(client);
                    }
                }
                else if (!wantsToProduceVideo && isCurrentlyProducingVideo) {
                    console.log(`[MEDIA PROXY AGENT] [${user_id}] Stopping video production.`);
                    producerClient.stopPublishingTrack("video");

                    for (const client of producerClient.room.clients.values()) {
                        if (client.user_id !== user_id) clientsThatNeedUpdate.add(client);
                    }
                }

                await Promise.all(
                    Array.from(clientsThatNeedUpdate).map((client) => {
                        const ssrcs = client.getOutgoingStreamSSRCsForUser(user_id);

                        video_batch[client.user_id] = {
                            op: 12,
                            d: {
                                user_id: user_id,
                                audio_ssrc: ssrcs.audio_ssrc,
                                video_ssrc: ssrcs.video_ssrc,
                                rtx_ssrc: ssrcs.rtx_ssrc
                            },
                        };
                    }),
                );

                if (Object.entries(video_batch).length > 0) {
                    socket.send(JSON.stringify({
                        op: "VIDEO_BATCH",
                        d: video_batch
                    }));
                }
            }
        });

        socket.on('close', () => {
            console.log(`[MEDIA PROXY AGENT] Main server disconnected!`);
            if (currentSocket === socket) {
                currentSocket = null;
                users.clear();
            }
        });

        socket.on('error', (err) => {
            console.log(`[MEDIA PROXY AGENT] Socket error: ${err}`);
        });
    });
}

start();