Rtms Bot

┌─────────────────────────────────────────────────────────────────────┐
│                        RTMS BOT FLOW                                 │
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│ 1. Trigger RTMS: REST API or In-Meeting Start                       │
│    └── POST /meetings/{meetingId}/rtms                              │
│    └── Or: Start RTMS manually from Zoom client                     │
└────────────────────────────┬────────────────────────────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────┐
│ 2. Wait for Webhook: meeting.rtms_started                           │
│    └── Zoom sends signaling + media WebSocket URLs                  │
│    └── No webhook = RTMS unavailable (no polling fallback)          │
└────────────────────────────┬────────────────────────────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────┐
│ 3. Connect: Signaling WebSocket (Handshake with HMAC)               │
│    └── Generate HMAC-SHA256 signature                               │
│    └── Send handshake message                                       │
│    └── Receive session confirmation                                 │
└────────────────────────────┬────────────────────────────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────┐
│ 4. Connect: Media WebSocket (Subscribe to Streams)                  │
│    └── Subscribe to: audio, video, transcription, share, chat       │
│    └── Send keep-alive pings                                        │
└────────────────────────────┬────────────────────────────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────┐
│ 5. Process Media Data                                               │
│    └── Audio: Opus/PCM streams per speaker                          │
│    └── Video: H.264 encoded frames                                  │
│    └── Transcription: Real-time text with speaker labels            │
└────────────────────────────┬────────────────────────────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────┐
│ 6. Mid-Stream: Connection Monitoring                                │
│    └── Detect WebSocket close → Exponential backoff retry           │
│    └── Stop after N reconnection attempts                           │
└─────────────────────────────────────────────────────────────────────┘

// config.js or environment variables
const rtmsConfig = {
    // WebSocket connection (initial)
    connection_timeout_ms: 10000,        // Handshake timeout (default: 10s)
    connection_max_attempts: 5,          // Max connection attempts (default: 5)
    connection_retry_delay_ms: 5000,     // Constant retry: 5s (default: 5s)
    
    // Mid-stream reconnection (network failures)
    reconnect_max_attempts: 3,           // Max reconnection attempts (default: 3)
    reconnect_base_delay_ms: 2000,       // Initial delay: 2s (default: 2s)
    // Exponential backoff: 2s, 4s, 8s...
    
    // Keep-alive ping
    keepalive_interval_ms: 5000,         // Send ping every 5s (default: 5s, min: 3s)
    keepalive_timeout_ms: 15000,         // Expect pong within 15s (default: 15s)
    
    // Webhook wait timeout
    webhook_wait_timeout_ms: 300000      // Wait 5min for webhook (default: 5min)
};

// Load from environment variables (recommended for production)
function loadConfig() {
    return {
        connection_timeout_ms: 
            parseInt(process.env.RTMS_CONNECTION_TIMEOUT_MS) || 10000,
        connection_max_attempts: 
            parseInt(process.env.RTMS_CONNECTION_MAX_ATTEMPTS) || 5,
        connection_retry_delay_ms: 
            parseInt(process.env.RTMS_CONNECTION_RETRY_DELAY_MS) || 5000,
        reconnect_max_attempts: 
            parseInt(process.env.RTMS_RECONNECT_MAX_ATTEMPTS) || 3,
        reconnect_base_delay_ms: 
            parseInt(process.env.RTMS_RECONNECT_BASE_DELAY_MS) || 2000,
        keepalive_interval_ms: 
            Math.max(parseInt(process.env.RTMS_KEEPALIVE_INTERVAL_MS) || 5000, 3000),
        keepalive_timeout_ms: 
            parseInt(process.env.RTMS_KEEPALIVE_TIMEOUT_MS) || 15000,
        webhook_wait_timeout_ms: 
            parseInt(process.env.RTMS_WEBHOOK_WAIT_TIMEOUT_MS) || 300000
    };
}

# High-priority production bot (aggressive)
export RTMS_CONNECTION_MAX_ATTEMPTS=10
export RTMS_CONNECTION_RETRY_DELAY_MS=3000      # 3s
export RTMS_RECONNECT_MAX_ATTEMPTS=5
export RTMS_RECONNECT_BASE_DELAY_MS=1000        # 1s
export RTMS_KEEPALIVE_INTERVAL_MS=3000          # 3s (minimum)

# Cost-sensitive batch processing (conservative)
export RTMS_CONNECTION_MAX_ATTEMPTS=3
export RTMS_CONNECTION_RETRY_DELAY_MS=10000     # 10s
export RTMS_RECONNECT_MAX_ATTEMPTS=2
export RTMS_RECONNECT_BASE_DELAY_MS=5000        # 5s
export RTMS_KEEPALIVE_INTERVAL_MS=15000         # 15s

# Development/testing (fail fast)
export RTMS_CONNECTION_MAX_ATTEMPTS=2
export RTMS_CONNECTION_RETRY_DELAY_MS=2000      # 2s
export RTMS_RECONNECT_MAX_ATTEMPTS=1
export RTMS_RECONNECT_BASE_DELAY_MS=1000        # 1s
export RTMS_WEBHOOK_WAIT_TIMEOUT_MS=60000       # 1min

# Start RTMS for a meeting
curl -X POST "https://api.zoom.us/v2/meetings/{meetingId}/rtms" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "meeting"
  }'

{
  "rtms_id": "abc123def456",
  "status": "starting"
}

const express = require('express');
const crypto = require('crypto');
const app = express();

const config = loadConfig();
const pendingConnections = new Map();  // Track webhook waiters

app.post('/webhook', express.json(), async (req, res) => {
    // 1. Verify webhook signature
    const signature = req.headers['x-zm-signature'];
    const timestamp = req.headers['x-zm-request-timestamp'];
    
    if (!verifyWebhookSignature(req.body, signature, timestamp)) {
        return res.status(403).send('Invalid signature');
    }
    
    // 2. Respond immediately (Zoom expects 200 within 3s)
    res.status(200).send();
    
    // 3. Process webhook asynchronously
    const event = req.body;
    
    if (event.event === 'meeting.rtms_started') {
        console.log('[WEBHOOK] RTMS started for meeting:', event.payload.object.uuid);
        
        const rtmsInfo = {
            meetingUuid: event.payload.object.uuid,
            signalingUrl: event.payload.object.signaling_url,
            mediaUrl: event.payload.object.media_url,
            sessionKey: event.payload.object.session_key
        };
        
        // Notify waiting connection
        const waiter = pendingConnections.get(rtmsInfo.meetingUuid);
        if (waiter) {
            waiter.resolve(rtmsInfo);
            pendingConnections.delete(rtmsInfo.meetingUuid);
        } else {
            // No waiter - proactive start
            connectToRTMS(rtmsInfo);
        }
    }
});

function verifyWebhookSignature(body, signature, timestamp) {
    const message = `v0:${timestamp}:${JSON.stringify(body)}`;
    const hmac = crypto.createHmac('sha256', WEBHOOK_SECRET_TOKEN);
    const computed = 'v0=' + hmac.update(message).digest('hex');
    return crypto.timingSafeEqual(
        Buffer.from(signature),
        Buffer.from(computed)
    );
}

async function waitForRTMSWebhook(meetingUuid) {
    return new Promise((resolve, reject) => {
        const timeoutId = setTimeout(() => {
            pendingConnections.delete(meetingUuid);
            reject(new Error(
                `RTMS webhook not received within ${config.webhook_wait_timeout_ms}ms. ` +
                `Possible causes: ` +
                `(1) Meeting hasn't started, ` +
                `(2) RTMS not enabled for this meeting, ` +
                `(3) Webhook endpoint unreachable.`
            ));
        }, config.webhook_wait_timeout_ms);
        
        pendingConnections.set(meetingUuid, {
            resolve: (rtmsInfo) => {
                clearTimeout(timeoutId);
                resolve(rtmsInfo);
            },
            reject
        });
    });
}

// Usage
try {
    console.log('[RTMS] Waiting for webhook...');
    const rtmsInfo = await waitForRTMSWebhook(MEETING_UUID);
    console.log('[RTMS] Webhook received, connecting...');
    await connectToRTMS(rtmsInfo);
} catch (error) {
    console.error('[RTMS] ABORT:', error.message);
}

const WebSocket = require('ws');

async function connectSignalingWithRetry(signalingUrl, sessionKey) {
    for (let attempt = 1; attempt <= config.connection_max_attempts; attempt++) {
        console.log(`[SIGNALING] Attempt ${attempt}/${config.connection_max_attempts}`);
        
        try {
            const ws = await connectSignalingSocket(signalingUrl, sessionKey);
            console.log('[SIGNALING] Connected successfully');
            return ws;
        } catch (error) {
            console.error(`[SIGNALING] Attempt ${attempt} failed:`, error.message);
            
            if (attempt < config.connection_max_attempts) {
                const delay = config.connection_retry_delay_ms;
                console.log(`[SIGNALING] Retrying in ${delay}ms...`);
                await sleep(delay);
            }
        }
    }
    
    throw new Error(
        `Failed to connect signaling WebSocket after ${config.connection_max_attempts} attempts`
    );
}

function connectSignalingSocket(signalingUrl, sessionKey) {
    return new Promise((resolve, reject) => {
        const ws = new WebSocket(signalingUrl);
        const timeoutId = setTimeout(() => {
            ws.close();
            reject(new Error('Signaling connection timeout'));
        }, config.connection_timeout_ms);
        
        ws.on('open', () => {
            console.log('[SIGNALING] WebSocket opened, sending handshake...');
            
            // Generate HMAC signature
            const timestamp = Date.now();
            const message = `${timestamp}:${sessionKey}`;
            const signature = crypto
                .createHmac('sha256', WEBHOOK_SECRET_TOKEN)
                .update(message)
                .digest('hex');
            
            // Send handshake
            ws.send(JSON.stringify({
                type: 'handshake',
                timestamp,
                signature
            }));
        });
        
        ws.on('message', (data) => {
            const msg = JSON.parse(data);
            
            if (msg.type === 'handshake_response') {
                clearTimeout(timeoutId);
                
                if (msg.status === 'success') {
                    console.log('[SIGNALING] Handshake successful');
                    resolve(ws);
                } else {
                    ws.close();
                    reject(new Error(`Handshake failed: ${msg.error}`));
                }
            }
        });
        
        ws.on('error', (error) => {
            clearTimeout(timeoutId);
            reject(error);
        });
        
        ws.on('close', (code, reason) => {
            clearTimeout(timeoutId);
            reject(new Error(`Connection closed: ${code} ${reason}`));
        });
    });
}

async function connectMediaWithRetry(mediaUrl, signalingWs) {
    for (let attempt = 1; attempt <= config.connection_max_attempts; attempt++) {
        console.log(`[MEDIA] Attempt ${attempt}/${config.connection_max_attempts}`);
        
        try {
            const ws = await connectMediaSocket(mediaUrl);
            console.log('[MEDIA] Connected successfully');
            subscribeToStreams(ws);
            setupKeepAlive(ws);
            return ws;
        } catch (error) {
            console.error(`[MEDIA] Attempt ${attempt} failed:`, error.message);
            
            if (attempt < config.connection_max_attempts) {
                const delay = config.connection_retry_delay_ms;
                console.log(`[MEDIA] Retrying in ${delay}ms...`);
                await sleep(delay);
            }
        }
    }
    
    throw new Error(
        `Failed to connect media WebSocket after ${config.connection_max_attempts} attempts`
    );
}

function subscribeToStreams(mediaWs) {
    // Subscribe to all available streams
    mediaWs.send(JSON.stringify({
        type: 'subscribe',
        streams: ['audio', 'video', 'transcription', 'share', 'chat']
    }));
    
    console.log('[MEDIA] Subscribed to: audio, video, transcription, share, chat');
}

function setupKeepAlive(ws) {
    let lastPongReceived = Date.now();
    let keepAliveInterval;
    let timeoutCheck;
    
    // Send ping periodically
    keepAliveInterval = setInterval(() => {
        if (ws.readyState === WebSocket.OPEN) {
            ws.ping();
            console.log('[KEEPALIVE] Ping sent');
        }
    }, config.keepalive_interval_ms);
    
    // Check for pong timeout
    timeoutCheck = setInterval(() => {
        const timeSinceLastPong = Date.now() - lastPongReceived;
        
        if (timeSinceLastPong > config.keepalive_timeout_ms) {
            console.error('[KEEPALIVE] Pong timeout, closing connection');
            clearInterval(keepAliveInterval);
            clearInterval(timeoutCheck);
            ws.close(1000, 'Keep-alive timeout');
        }
    }, 1000);
    
    ws.on('pong', () => {
        lastPongReceived = Date.now();
        console.log('[KEEPALIVE] Pong received');
    });
    
    ws.on('close', () => {
        clearInterval(keepAliveInterval);
        clearInterval(timeoutCheck);
    });
}

class ResilientRTMSConnection {
    constructor(rtmsInfo, config) {
        this.rtmsInfo = rtmsInfo;
        this.config = config;
        this.reconnectionAttempt = 0;
        this.signalingWs = null;
        this.mediaWs = null;
    }
    
    async connect() {
        try {
            this.signalingWs = await connectSignalingWithRetry(
                this.rtmsInfo.signalingUrl,
                this.rtmsInfo.sessionKey
            );
            
            this.mediaWs = await connectMediaWithRetry(
                this.rtmsInfo.mediaUrl,
                this.signalingWs
            );
            
            this.setupReconnectionHandlers();
            
        } catch (error) {
            console.error('[RTMS] Initial connection failed:', error);
            throw error;
        }
    }
    
    setupReconnectionHandlers() {
        const handleDisconnection = (wsType) => async (code, reason) => {
            console.error(`[${wsType}] Disconnected: ${code} ${reason}`);
            
            this.reconnectionAttempt++;
            
            if (this.reconnectionAttempt > this.config.reconnect_max_attempts) {
                console.error(
                    `[RECONNECT] Giving up after ${this.reconnectionAttempt} attempts`
                );
                this.cleanup();
                return;
            }
            
            // Exponential backoff: 2s, 4s, 8s...
            const delay = this.config.reconnect_base_delay_ms 
                          * Math.pow(2, this.reconnectionAttempt - 1);
            
            console.log(
                `[RECONNECT] Attempt ${this.reconnectionAttempt}/` +
                `${this.config.reconnect_max_attempts} in ${delay}ms...`
            );
            
            await sleep(delay);
            
            try {
                await this.connect();
                console.log('[RECONNECT] Successfully reconnected');
                this.reconnectionAttempt = 0;  // Reset counter
            } catch (error) {
                console.error('[RECONNECT] Failed:', error.message);
                // Handler will be called again if connection fails
            }
        };
        
        this.signalingWs.on('close', handleDisconnection('SIGNALING'));
        this.mediaWs.on('close', handleDisconnection('MEDIA'));
        
        this.signalingWs.on('error', (error) => {
            console.error('[SIGNALING] Error:', error.message);
        });
        
        this.mediaWs.on('error', (error) => {
            console.error('[MEDIA] Error:', error.message);
        });
    }
    
    cleanup() {
        if (this.signalingWs) this.signalingWs.close();
        if (this.mediaWs) this.mediaWs.close();
    }
}

// Example: Capped exponential backoff (max 30s)
const delay = Math.min(
    config.reconnect_base_delay_ms * Math.pow(2, reconnectionAttempt - 1),
    30000  // Cap at 30s
);

// Example: Linear backoff instead of exponential
const delay = config.reconnect_base_delay_ms * reconnectionAttempt;

// Example: Jittered backoff (avoid thundering herd)
const baseDelay = config.reconnect_base_delay_ms * Math.pow(2, reconnectionAttempt - 1);
const jitter = Math.random() * 1000;  // Random 0-1000ms
const delay = baseDelay + jitter;

const config = loadConfig();

async function main() {
    try {
        // 1. Optional: Trigger RTMS via REST API
        console.log('[RTMS] Triggering RTMS start...');
        await triggerRTMSStart(MEETING_ID);
        
        // 2. Wait for webhook
        console.log('[RTMS] Waiting for meeting.rtms_started webhook...');
        const rtmsInfo = await waitForRTMSWebhook(MEETING_UUID);
        
        // 3. Connect with resilience
        const rtms = new ResilientRTMSConnection(rtmsInfo, config);
        await rtms.connect();
        
        console.log('[RTMS] Bot is running, processing streams...');
        
        // 4. Process media data
        rtms.mediaWs.on('message', (data) => {
            const frame = parseMediaFrame(data);
            processMediaFrame(frame);
        });
        
        // 5. Handle graceful shutdown
        process.on('SIGINT', () => {
            console.log('[RTMS] Shutting down...');
            rtms.cleanup();
            process.exit(0);
        });
        
    } catch (error) {
        console.error('[RTMS] ABORT:', error.message);
        process.exit(1);
    }
}

main();