feat(probe): pre-flight reachability + actionable SRT/RTMP error messages

services/capture/src/routes/capture.js

@@ -2,6 +2,77 @@ import express from 'express';
 import { execSync, spawn } from 'child_process';
 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';
@@ -62,7 +133,7 @@ router.get('/status', (req, res) => {
 });
 router.post('/probe', async (req, res) => {
   try {
-    const { source_type = 'sdi', source_url, listen = false, device } = req.body || {};
+    const { source_type = 'sdi', source_url, listen = false } = req.body || {};
 
     if (source_type === 'sdi') {
       try {
@@ -85,6 +156,17 @@ router.post('/probe', async (req, res) => {
 
     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';
@@ -99,7 +181,9 @@ router.post('/probe', async (req, res) => {
     ff.on('close', (code) => {
       clearTimeout(killer);
       if (code !== 0) {
-        return res.json({ ok: false, source_type, source_url, error: (stderr || 'ffprobe failed').slice(0, 800) });
+        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);
@@ -123,7 +207,6 @@ router.post('/probe', async (req, res) => {
   }
 });
 
-
 /**
  * POST /start
  * Start a new capture session