← back to Butlr
lib/listen-bridge.js
261 lines
// Live-audio listen bridge.
//
// Twilio Voice → <Start><Stream url="wss://.../stream/<call_id>" track="both_tracks"/>
// sends µ-law audio frames at 8kHz mono in 20ms chunks as base64 JSON
// envelopes: { event: "media", media: { track: "inbound"|"outbound", payload: "<base64>" } }.
//
// With track="both_tracks" the two halves of the conversation arrive as
// SEPARATE frames distinguished by media.track — they are NOT pre-mixed.
// We archive each track into its own .pcm file and ffmpeg-merge to a
// STEREO MP3 at call end (inbound=left, outbound=right). Live broadcast
// to /listen-ws sends BOTH tracks (each 20ms frame as it arrives) so a
// browser listener hears the full conversation; the half-duplex nature
// of a phone call makes arrival-order interleave sound clean (no client
// change needed — still mono 8kHz Int16 frames).
//
// Two WebSocket "lanes" on the same upgrade handler — Twilio Streams on
// /stream/:call_id, browser listeners on /listen-ws/:call_id.
const { WebSocketServer } = require('ws');
const url = require('url');
const fs = require('fs');
const path = require('path');
const { spawn } = require('child_process');
// ── Per-call audio archive ──────────────────────────────────────────────
// Each call gets two raw-PCM files (inbound + outbound), ffmpeg-merged
// into a stereo MP3 on close. Persists to disk regardless of whether
// any browser is listening — the bug that lost the 2026-05-12 Wells
// Fargo call (DTCI7PuZ).
const RECORDINGS_DIR = path.join(__dirname, '..', 'data', 'recordings');
fs.mkdirSync(RECORDINGS_DIR, { recursive: true });
const pcmStreams = new Map(); // callId → handle{ inboundFd, outboundFd, paths, bytes }
// ── Orphan-PCM recovery on boot ─────────────────────────────────────
// If the process died mid-call (SIGINT from zombie pm2 God, OOM, etc.),
// the WS close handler never fired and PCM files sit unconverted on
// disk. Scan recordings/ at module-load time and ffmpeg-merge:
// • paired <id>.inbound.pcm + <id>.outbound.pcm → stereo .mp3
// • lone <id>.inbound.pcm OR <id>.outbound.pcm → mono .mp3
// • legacy <id>.pcm (pre-2026-05-21 schema) → mono .mp3
// Best-effort — failures are logged, not thrown.
(function recoverOrphanPcms() {
let entries;
try { entries = fs.readdirSync(RECORDINGS_DIR); } catch { return; }
const byCall = new Map(); // callId → { inbound?:path, outbound?:path, legacy?:path }
for (const f of entries) {
const inMatch = f.match(/^([A-Za-z0-9_-]{6,16})\.inbound\.pcm$/);
const outMatch = f.match(/^([A-Za-z0-9_-]{6,16})\.outbound\.pcm$/);
const oldMatch = f.match(/^([A-Za-z0-9_-]{6,16})\.pcm$/);
if (inMatch) (byCall.get(inMatch[1]) || byCall.set(inMatch[1], {}).get(inMatch[1])).inbound = path.join(RECORDINGS_DIR, f);
else if (outMatch) (byCall.get(outMatch[1]) || byCall.set(outMatch[1], {}).get(outMatch[1])).outbound = path.join(RECORDINGS_DIR, f);
else if (oldMatch) (byCall.get(oldMatch[1]) || byCall.set(oldMatch[1], {}).get(oldMatch[1])).legacy = path.join(RECORDINGS_DIR, f);
}
if (!byCall.size) return;
console.log(`[listen-bridge] orphan-PCM recovery: ${byCall.size} unfinalized recording(s)`);
for (const [callId, parts] of byCall) {
const mp3Path = path.join(RECORDINGS_DIR, `${callId}.mp3`);
if (fs.existsSync(mp3Path)) {
for (const p of Object.values(parts)) { try { fs.unlinkSync(p); } catch {} }
continue;
}
runFfmpegMerge(callId, parts, mp3Path, /*sourceHint*/'orphan');
}
})();
// runFfmpegMerge handles all three input shapes:
// { inbound, outbound } → stereo (inbound=L, outbound=R)
// { inbound } → mono
// { outbound } → mono
// { legacy } → mono (pre-stereo schema)
function runFfmpegMerge(callId, parts, mp3Path, sourceHint) {
const hasIn = !!parts.inbound && safeSize(parts.inbound) >= 4096;
const hasOut = !!parts.outbound && safeSize(parts.outbound) >= 4096;
const hasOld = !!parts.legacy && safeSize(parts.legacy) >= 4096;
if (!hasIn && !hasOut && !hasOld) {
// All under-size or empty — drop everything, no MP3.
for (const p of Object.values(parts)) { try { fs.unlinkSync(p); } catch {} }
return;
}
let args;
let mode;
if (hasIn && hasOut) {
// Stereo: inbound left, outbound right.
args = [
'-y',
'-f','s16le','-ar','8000','-ac','1','-i', parts.inbound,
'-f','s16le','-ar','8000','-ac','1','-i', parts.outbound,
'-filter_complex', '[0:a][1:a]amerge=inputs=2[a]',
'-map','[a]','-ac','2',
'-codec:a','libmp3lame','-q:a','4',
mp3Path,
];
mode = 'stereo';
} else {
const monoPath = hasIn ? parts.inbound : (hasOut ? parts.outbound : parts.legacy);
args = ['-y','-f','s16le','-ar','8000','-ac','1','-i', monoPath,
'-codec:a','libmp3lame','-q:a','4', mp3Path];
mode = hasIn ? 'mono(in)' : (hasOut ? 'mono(out)' : 'mono(legacy)');
}
const ff = spawn('ffmpeg', args, { stdio: 'ignore' });
ff.on('exit', (code) => {
if (code !== 0) {
console.error(`[listen-bridge] ffmpeg exit ${code} for call=${callId} (${sourceHint}, ${mode})`);
return;
}
for (const p of Object.values(parts)) { try { fs.unlinkSync(p); } catch {} }
const sIn = parts.inbound ? safeSize(parts.inbound) : 0;
const sOut = parts.outbound ? safeSize(parts.outbound) : 0;
console.log(`[listen-bridge] archived ${mode} ${sourceHint} call=${callId} → ${path.basename(mp3Path)} (in:${sIn} out:${sOut})`);
try {
const t = require('./transcribe');
if (t && t.transcribe) t.transcribe(callId, mp3Path).then(r => {
if (r && r.ok) console.log(`[listen-bridge] auto-transcribed call=${callId} (source=${r.source})`);
}).catch(() => {});
} catch {}
});
}
function safeSize(p) { try { return fs.statSync(p).size; } catch { return 0; } }
function openArchive(callId) {
if (pcmStreams.has(callId)) return pcmStreams.get(callId);
const inboundPath = path.join(RECORDINGS_DIR, `${callId}.inbound.pcm`);
const outboundPath = path.join(RECORDINGS_DIR, `${callId}.outbound.pcm`);
const inboundFd = fs.openSync(inboundPath, 'a');
const outboundFd = fs.openSync(outboundPath, 'a');
const handle = { inboundFd, outboundFd, inboundPath, outboundPath, inboundBytes: 0, outboundBytes: 0 };
pcmStreams.set(callId, handle);
return handle;
}
function archiveAppend(callId, track, pcmBuf) {
const h = openArchive(callId);
try {
if (track === 'outbound') { fs.writeSync(h.outboundFd, pcmBuf); h.outboundBytes += pcmBuf.length; }
else { fs.writeSync(h.inboundFd, pcmBuf); h.inboundBytes += pcmBuf.length; }
} catch {}
}
function finalizeArchive(callId) {
const h = pcmStreams.get(callId);
if (!h) return;
pcmStreams.delete(callId);
try { fs.closeSync(h.inboundFd); } catch {}
try { fs.closeSync(h.outboundFd); } catch {}
const mp3Path = path.join(RECORDINGS_DIR, `${callId}.mp3`);
runFfmpegMerge(callId, { inbound: h.inboundPath, outbound: h.outboundPath }, mp3Path, /*sourceHint*/'live');
}
// µ-law → linear PCM16 LE decoder (G.711 standard). Bit-fiddly but small.
// G.711 decode bias is 0x84 (132) and is added INSIDE the << exponent shift.
// This was 33 — under-counted by 4× — which left quiet samples (exponent 0)
// correct but nonlinearly crushed every loud sample (e.g. 0x00 decoded to
// -19551 instead of -32124). That amplitude-dependent distortion is what
// made the recordings and the live-listen feed sound horrible.
const MULAW_BIAS = 0x84;
function mulawDecodeOne(mulawByte) {
mulawByte = ~mulawByte & 0xff;
const sign = (mulawByte & 0x80) ? -1 : 1;
const exponent = (mulawByte >> 4) & 0x07;
const mantissa = mulawByte & 0x0f;
let sample = ((mantissa << 3) + MULAW_BIAS) << exponent;
sample -= MULAW_BIAS;
return sign * sample;
}
function mulawDecodeBuffer(mulawBuf) {
const out = Buffer.alloc(mulawBuf.length * 2);
for (let i = 0; i < mulawBuf.length; i++) {
out.writeInt16LE(mulawDecodeOne(mulawBuf[i]), i * 2);
}
return out;
}
// callId → Set<browserWs>
const listeners = new Map();
function getListeners(callId) {
let set = listeners.get(callId);
if (!set) { set = new Set(); listeners.set(callId, set); }
return set;
}
function broadcastPcm(callId, pcmBuf) {
const set = listeners.get(callId);
if (!set || set.size === 0) return;
for (const ws of set) {
if (ws.readyState === ws.OPEN) {
try { ws.send(pcmBuf, { binary: true }); } catch (_) { /* swallow */ }
}
}
}
function attachListenBridge(httpServer) {
const wss = new WebSocketServer({ noServer: true });
httpServer.on('upgrade', (req, socket, head) => {
const parsed = url.parse(req.url);
const m = (parsed.pathname || '').match(/^\/(stream|listen-ws)\/([A-Za-z0-9_-]{6,16})$/);
if (!m) { socket.destroy(); return; }
const lane = m[1];
const callId = m[2];
// Defense-in-depth: only Twilio may open the /stream lane.
if (lane === 'stream') {
const ua = String(req.headers['user-agent'] || '');
if (!/^Twilio[A-Za-z.]*\/[0-9]/.test(ua)) {
console.error('[listen-bridge] rejected /stream upgrade — bad UA:', ua.slice(0, 80));
socket.destroy();
return;
}
}
wss.handleUpgrade(req, socket, head, (ws) => {
ws._lane = lane;
ws._callId = callId;
if (lane === 'listen-ws') {
getListeners(callId).add(ws);
ws.on('close', () => {
const set = listeners.get(callId);
if (set) { set.delete(ws); if (set.size === 0) listeners.delete(callId); }
});
ws.on('error', () => {});
return;
}
// lane === 'stream' (Twilio Media Streams)
let msgCount = 0;
let nonMediaCount = 0;
let mediaCount = 0;
ws.on('message', (data, isBinary) => {
msgCount++;
if (msgCount <= 3) {
// Log the first 3 raw messages so we can see what Twilio is actually sending.
console.log(`[listen-bridge] stream/${callId} msg#${msgCount} isBinary=${isBinary} len=${data.length} preview=${data.toString('utf8').slice(0,200)}`);
}
if (isBinary) { console.log(`[listen-bridge] stream/${callId} ignored binary msg #${msgCount}`); return; }
let msg;
try { msg = JSON.parse(data.toString('utf8')); } catch (e) { console.log(`[listen-bridge] stream/${callId} parse fail: ${e.message}`); return; }
if (msg.event !== 'media') { nonMediaCount++; if (nonMediaCount <= 5) console.log(`[listen-bridge] stream/${callId} non-media event: ${msg.event}`); }
if (msg.event !== 'media' || !msg.media || !msg.media.payload) return;
mediaCount++;
const track = (msg.media.track === 'outbound') ? 'outbound' : 'inbound';
const mulaw = Buffer.from(msg.media.payload, 'base64');
const pcm = mulawDecodeBuffer(mulaw);
archiveAppend(callId, track, pcm); // both tracks → separate .pcm files
// Live listen broadcasts BOTH tracks (not just inbound) so Steve hears the full
// conversation — the AI speaks on outbound, the business on inbound. We send each
// 20ms frame as it arrives rather than per-sample summing: a phone call is mostly
// half-duplex (one party at a time), so arrival-order interleave in the browser's
// nextPlayTime queue reproduces the conversation cleanly with no clipping/buffering
// latency, and needs zero client change (it already expects mono 8kHz Int16 frames).
broadcastPcm(callId, pcm);
});
ws.on('close', () => { finalizeArchive(callId); });
ws.on('error', () => { finalizeArchive(callId); });
});
});
}
module.exports = { attachListenBridge, getListeners, mulawDecodeBuffer };