import express from 'express'; import { execSync, spawn } from 'child_process'; import { existsSync, readdirSync } from 'node:fs'; import captureManager from '../capture-manager.js'; import dgram from 'dgram'; import net from 'net'; function parseUrl(u) { try { const m = String(u).match(/^[a-z]+:\/\/([^:\/?#]+)(?::(\d+))?/i); if (!m) return null; return { host: m[1], port: parseInt(m[2] || '0', 10) }; } catch (_) { return null; } } async function checkReachable(host, port, sourceType) { if (!port) return { ok: true }; if (sourceType === 'srt') return await udpSendProbe(host, port); if (sourceType === 'rtmp') return await tcpConnectProbe(host, port); return { ok: true }; } function udpSendProbe(host, port) { return new Promise((resolve) => { const sock = dgram.createSocket('udp4'); let done = false; const finish = (result) => { if (done) return; done = true; try { sock.close(); } catch (_) {} resolve(result); }; sock.on('error', (err) => { const msg = String(err && err.message || err); if (/EHOSTUNREACH|ENETUNREACH/i.test(msg)) { finish({ ok: false, error: 'Host ' + host + ' is unreachable from the capture container (no route). Confirm the IP is correct and the machine is online.', diagnostic: msg }); } else if (/ECONNREFUSED|EPORTUNREACH/i.test(msg)) { finish({ ok: false, error: 'Nothing is listening on UDP ' + host + ':' + port + '. In vMix, confirm the SRT output is Started and the port matches.', diagnostic: msg }); } else { finish({ ok: false, error: 'UDP probe failed: ' + msg, diagnostic: msg }); } }); sock.send(Buffer.from('Z-AMPP-PROBE'), port, host, () => {}); setTimeout(() => finish({ ok: true }), 1500); }); } function tcpConnectProbe(host, port) { return new Promise((resolve) => { const sock = new net.Socket(); let done = false; const finish = (r) => { if (done) return; done = true; try { sock.destroy(); } catch (_) {} resolve(r); }; sock.setTimeout(2500); sock.once('connect', () => finish({ ok: true })); sock.once('timeout', () => finish({ ok: false, error: 'TCP connect to ' + host + ':' + port + ' timed out. Confirm the host is reachable and a TCP listener is running.' })); sock.once('error', (err) => { const msg = String(err && err.message || err); if (/EHOSTUNREACH|ENETUNREACH/i.test(msg)) finish({ ok: false, error: 'Host ' + host + ' unreachable (no route).', diagnostic: msg }); else if (/ECONNREFUSED/i.test(msg)) finish({ ok: false, error: 'Nothing is listening on TCP ' + host + ':' + port + '.', diagnostic: msg }); else finish({ ok: false, error: 'TCP probe failed: ' + msg, diagnostic: msg }); }); sock.connect(port, host); }); } function classifyProbeError(raw, sourceType) { const r = (raw || '').toLowerCase(); if (sourceType === 'srt') { if (/connection .* failed: (input\/output|timer expired|connection setup failure)/i.test(raw)) { return 'SRT handshake failed. In vMix: confirm the External Output is Started, Type=SRT, Mode=Listener, port matches, and any passphrase / stream ID is empty (or copied exactly).'; } } if (sourceType === 'rtmp') { if (/connection refused/i.test(r)) return 'Nothing is listening on RTMP at this address. Start your RTMP source.'; if (/end-of-file|invalid data found/i.test(r)) return 'Got a TCP connection but no RTMP stream. Confirm the source is publishing and the path / stream-key match.'; } return raw; } const router = express.Router(); const MAM_API_URL = process.env.MAM_API_URL || 'http://mam-api:3000'; /** * GET /devices * List available DeckLink devices */ router.get('/devices', (req, res) => { try { const devices = []; let output = ''; try { output = execSync('ffmpeg -sources decklink 2>&1', { encoding: 'utf-8', }); } catch (error) { // ffmpeg returns non-zero, but stderr is still captured output = error.stderr ? error.stderr.toString() : error.toString(); } // Parse ffmpeg output for DeckLink device names. // DeckLink source lines: " 81:76669a80:00000000 [DeckLink Duo (1)] (none)" const lines = output.split('\n'); let deviceIndex = 0; for (const line of lines) { const match = line.match(/^\s+[0-9a-f:]+\s+\[([^\]]+)\]/); if (match) { devices.push({ index: deviceIndex, name: match[1], }); deviceIndex++; } } res.json({ devices }); } catch (error) { console.error('Error listing devices:', error); res.status(500).json({ error: 'Failed to list devices' }); } }); /** * GET /devices/deltacast * List available Deltacast ports. * Reads /dev/deltacast nodes; falls back to env DELTACAST_PORT_COUNT * so nodes without hardware still report their configured port count * (test-card mode). */ router.get('/devices/deltacast', (req, res) => { try { const devices = []; // First: enumerate actual /dev/deltacast* device nodes. try { const devEntries = readdirSync('/dev').filter(n => /^deltacast\d+$/.test(n)); devEntries.sort(); for (const entry of devEntries) { const m = entry.match(/^deltacast(\d+)$/); if (m) { devices.push({ index: parseInt(m[1], 10), name: `Deltacast Port ${m[1]}`, device: `/dev/${entry}`, present: true, }); } } } catch (_) { /* /dev always exists; ignore */ } // Second: if DELTACAST_PORT_COUNT env is set and larger than what we found, // fill in the remaining slots as test-card entries (no physical device). const envCount = parseInt(process.env.DELTACAST_PORT_COUNT || '0', 10); const found = new Set(devices.map(d => d.index)); for (let i = 0; i < envCount; i++) { if (!found.has(i)) { devices.push({ index: i, name: `Deltacast Port ${i} (test card)`, device: `/dev/deltacast${i}`, present: false, }); } } devices.sort((a, b) => a.index - b.index); res.json({ devices }); } catch (error) { console.error('Error listing Deltacast devices:', error); res.status(500).json({ error: 'Failed to list Deltacast devices' }); } }); /** * GET /status * Get current capture status */ router.get('/status', (req, res) => { try { const status = captureManager.getStatus(); res.json(status); } catch (error) { console.error('Error getting status:', error); res.status(500).json({ error: 'Failed to get status' }); } }); router.post('/probe', async (req, res) => { try { const { source_type = 'sdi', source_url, listen = false } = req.body || {}; if (source_type === 'sdi') { try { const raw = execSync('ffmpeg -hide_banner -sources decklink 2>&1', { encoding: 'utf-8', timeout: 5000 }); const devices = []; for (const line of raw.split('\n')) { const m = line.match(/^\s+[0-9a-f:]+\s+\[([^\]]+)\]/); if (m) devices.push(m[1]); } return res.json({ ok: true, source_type, devices }); } catch (err) { const out = (err.stderr || err.stdout || err.toString()).toString(); return res.json({ ok: false, source_type, error: out.slice(0, 800) }); } } if (source_type === 'deltacast') { // Enumerate /dev/deltacast* nodes; report present/absent per index. try { const envCount = parseInt(process.env.DELTACAST_PORT_COUNT || '0', 10); const devEntries = readdirSync('/dev').filter(n => /^deltacast\d+$/.test(n)).sort(); const found = devEntries.map(n => { const m = n.match(/^deltacast(\d+)$/); return { index: parseInt(m[1], 10), device: `/dev/${n}`, present: true }; }); const foundIdx = new Set(found.map(d => d.index)); for (let i = 0; i < envCount; i++) { if (!foundIdx.has(i)) { found.push({ index: i, device: `/dev/deltacast${i}`, present: false }); } } found.sort((a, b) => a.index - b.index); return res.json({ ok: true, source_type, devices: found }); } catch (err) { return res.json({ ok: false, source_type, error: err.message }); } } if (listen) { return res.json({ ok: false, source_type, error: 'Listener-mode sources cannot be probed standalone. Start the recorder and watch the signal indicator.' }); } if (!source_url) return res.status(400).json({ error: 'source_url is required' }); // Pre-flight: parse host:port and check L3/L4 reachability so we can give // an actionable error instead of the opaque libsrt "Input/output error". const parsed = parseUrl(source_url); if (!parsed) { return res.json({ ok: false, source_type, source_url, error: 'Could not parse host:port from URL.' }); } const reach = await checkReachable(parsed.host, parsed.port, source_type); if (!reach.ok) { return res.json({ ok: false, source_type, source_url, error: reach.error, diagnostic: reach.diagnostic }); } let url = source_url; if (source_type === 'srt' && !/mode=/.test(url)) { url += (url.includes('?') ? '&' : '?') + 'mode=caller'; } const args = ['-hide_banner','-v','error','-probesize','32M','-analyzeduration','8M','-rw_timeout','7000000','-i', url, '-show_streams','-show_format','-of','json']; const ff = spawn('ffprobe', args); let stdout = '', stderr = ''; ff.stdout.on('data', (c) => { stdout += c; }); ff.stderr.on('data', (c) => { stderr += c; }); const killer = setTimeout(() => { try { ff.kill('SIGKILL'); } catch (_) {} }, 10000); ff.on('close', (code) => { clearTimeout(killer); if (code !== 0) { const rawErr = (stderr || 'ffprobe failed').slice(0, 800); const friendly = classifyProbeError(rawErr, source_type); return res.json({ ok: false, source_type, source_url, error: friendly, diagnostic: rawErr }); } try { const parsed = JSON.parse(stdout); const streams = (parsed.streams || []).map(s => ({ index: s.index, codec_type: s.codec_type, codec_name: s.codec_name, width: s.width, height: s.height, pix_fmt: s.pix_fmt, r_frame_rate: s.r_frame_rate, avg_frame_rate: s.avg_frame_rate, sample_rate: s.sample_rate, channels: s.channels, channel_layout: s.channel_layout, bit_rate: s.bit_rate, })); return res.json({ ok: true, source_type, source_url, format: { format_name: parsed.format && parsed.format.format_name, duration: parsed.format && parsed.format.duration, bit_rate: parsed.format && parsed.format.bit_rate }, streams }); } catch (err) { return res.json({ ok: false, source_type, source_url, error: 'Could not parse ffprobe output: ' + err.message }); } }); } catch (error) { console.error('Probe error:', error); res.status(500).json({ error: error.message }); } }); /** * POST /start * Start a new capture session * * Body (SDI): * { project_id, clip_name, device, bin_id?, source_type? } * * Body (SRT/RTMP caller): * { project_id, clip_name, source_type, source_url, bin_id? } * * Body (SRT/RTMP listener): * { project_id, clip_name, source_type, listen: true, listen_port?, stream_key?, bin_id? } */ router.post('/start', async (req, res) => { try { const { project_id, bin_id, clip_name, device, source_type = 'sdi', source_url, listen = false, listen_port, stream_key, } = req.body; if (!project_id || !clip_name) { return res.status(400).json({ error: 'Missing required fields: project_id, clip_name', }); } // Source-specific validation if (source_type === 'sdi') { if (device === undefined || device === null) { return res.status(400).json({ error: 'SDI source requires: device' }); } } else if (source_type === 'srt' || source_type === 'rtmp') { if (!listen && !source_url) { return res.status(400).json({ error: `${source_type.toUpperCase()} caller mode requires: source_url`, }); } } else { return res.status(400).json({ error: `Unknown source_type: ${source_type}. Must be sdi, srt, or rtmp`, }); } const session = await captureManager.start({ projectId: project_id, binId: bin_id || null, clipName: clip_name, device, sourceType: source_type, sourceUrl: source_url, listen, listenPort: listen_port, streamKey: stream_key, }); res.json(session); } catch (error) { console.error('Error starting capture:', error); res.status(500).json({ error: error.message }); } }); /** * POST /stop * Stop the current capture session * Body: { session_id } */ router.post('/stop', async (req, res) => { try { const { session_id } = req.body; if (!session_id) { return res.status(400).json({ error: 'Missing required field: session_id' }); } const completedSession = await captureManager.stop(session_id); // Register asset with mam-api. // If proxyKey is null (SRT/RTMP source), set needsProxy=true so the // worker generates a proxy from the hires file asynchronously. try { const mamResponse = await fetch(`${MAM_API_URL}/api/v1/assets`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ projectId: completedSession.projectId, binId: completedSession.binId, clipName: completedSession.clipName, sourceType: completedSession.sourceType, hiresKey: completedSession.hiresKey, proxyKey: completedSession.proxyKey, needsProxy: completedSession.proxyKey === null, duration: completedSession.duration, capturedAt: completedSession.startedAt, }), }); if (!mamResponse.ok) { console.warn( `MAM API registration returned ${mamResponse.status}: ${await mamResponse.text()}`, ); } } catch (mamError) { console.warn('Failed to register asset with MAM API:', mamError.message); } res.json(completedSession); } catch (error) { console.error('Error stopping capture:', error); res.status(500).json({ error: error.message }); } }); export default router;