diff --git a/AGENTS.md b/AGENTS.md index ef95a94..0c73757 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,7 +74,15 @@ Set these env vars (the server reads **these exact names** — note the checked- ### Phase 4 — Deploy 🧍 **Human required:** provide a public HTTPS host (or tunnel) for `PUBLIC_BASE_URL`. The BW Voice Application's callback (set in Phase 2) must point -at `/bw/initiate`. +at `/bw/initiate`. If the customer app uses ``, +the host must also pass WebSocket upgrades through to the translator at +`/bw/stream` (ngrok and cloudflared do by default). Optional stream tuning, all +in ms: `STREAM_PLAYOUT_LATENCY_PAD_MS` delays mark acknowledgements to the bot +to absorb network and jitter-buffer latency (default 0); +`STREAM_START_TIMEOUT_MS` bounds the wait for Bandwidth's `start` event +(default 5000); `STREAM_BOT_CONNECT_TIMEOUT_MS` bounds the bot's WebSocket +handshake (default 10000). Either timeout ends the Bandwidth stream so the +call's BXML moves on. ### Phase 5 — Verify ```bash @@ -112,8 +120,17 @@ Translation is a fixed rulebook (`src/matrix/twilio-voice.json`), not a guess. after the stream ends, as on Twilio. A stream name is generated when the TwiML omits one. `ConversationRelay` and `VirtualAgent` are unsupported (separate IoV). - - `Stream` — Twilio's WS message schema is emulated by the translator's stream - bridge; live Bandwidth-side binding requires fixture capture. `` + - `Stream` — the translator rewrites the Twilio `` (the customer's + bot) to `wss:///bw/stream?dest=`, so Bandwidth's + StartStream WebSocket lands on the translator, which bridges it to the bot in + Twilio's Media Streams protocol (`src/streams/bw-source.ts` speaks Bandwidth's + side, `src/streams/bridge.ts` Twilio's). The upgrade requires the same + Basic-auth credentials as the `/bw/*` webhooks, stamped on `StartStream` as + `destinationUsername`/`destinationPassword`, and `dest` passes the egress + guard. Only the inbound (caller) track is relayed; Bandwidth delivers no DTMF + over the media socket. With `TRANSLATOR_CAPTURE_DIR` set, raw Bandwidth + frames are written to `/streams/*.jsonl` (start, first 10 media, stop) + to refresh `test/fixtures/bandwidth/stream-frames.json`. `` children map to nested `` elements in order (Bandwidth allows at most 12; extras are dropped with a warning). Bandwidth echoes them in its `start` event as `streamParams`, and the bridge forwards them to the bot as diff --git a/README.md b/README.md index abb131e..e4dfe02 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,8 @@ real calls. | Record (+ `recordingStatusCallback`) | ✅ | Live | | Dial → Number / SIP | ✅ | Live | | Dial → Conference (mute + events; no hold music / lifecycle) | ⚠️ | Live | -| Connect/Start/Stop Stream, Start/Stop Transcription | ✅ | Unit | +| Connect → Stream (bidirectional, via the Media Streams bridge) | ✅ | Live | +| Start/Stop Stream (fork), Start/Stop Transcription | ✅ | Unit | | Reject, Refer (SIP) | ⚠️ | Unit | | Queue, Client, Enqueue, Leave, Pay | ❌ | — (no BW primitive — fail loudly) | @@ -128,7 +129,7 @@ real calls. | List / fetch / download recordings; pause-resume | ✅ | Live/Unit | | Status callback (call completion) | ✅ | Unit | | Recording callback (`recordingStatusCallback`) | ✅ | Unit | -| Media Streams bridge (AI-voice path) | ✅ | Unit | +| Media Streams bridge (AI-voice path) | ✅ | Live | Number provisioning (search/order/activate) isn't part of this toolkit — it's handled by the [`band` CLI](#pairs-with-the-band-cli); see the Phase 2 runbook @@ -198,7 +199,7 @@ All translation is driven by one declarative compatibility matrix (`src/matrix/t ``` src/translator TwiML → BXML translation src/twilio Twilio-shaped REST facade + signed webhooks (egress) -src/streams Media Streams bridge +src/streams Media Streams bridge (Twilio protocol to the bot) + Bandwidth StartStream source src/server the proxy (Fastify) wiring it together, readiness check (/readyz, npm run doctor) src/compatibility-check static migration-complexity report (npm run compatibility-check) src/bxml-generator batch BXML generation + coverage report (npm run bxml-generator; see AGENTS.md Phase 1) diff --git a/src/server/app.ts b/src/server/app.ts index 3974d95..fc36ec1 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -1,6 +1,12 @@ import Fastify, { type FastifyInstance, type FastifyReply, type FastifyRequest } from "fastify"; import formbody from "@fastify/formbody"; +import type { IncomingMessage } from "node:http"; +import type { Duplex } from "node:stream"; +import { randomUUID } from "node:crypto"; +import { WebSocketServer } from "ws"; import { translateTwiml, type UrlKind } from "../translator/translate.js"; +import { TwilioStreamBridge, customParametersFromBwStart } from "../streams/bridge.js"; +import { BwWebSocketSource } from "../streams/bw-source.js"; import { bxmlDocument } from "../xml/build-xml.js"; import { initiateParams, @@ -9,7 +15,7 @@ import { recordingStatusParams, postToCustomer, } from "../twilio/egress.js"; -import { EgressBlockedError } from "../twilio/egress-guard.js"; +import { EgressBlockedError, assertPublicUrl } from "../twilio/egress-guard.js"; import { toCallSid, toRecordingSid } from "../twilio/call-sid.js"; import { createdCallResource, bwStateToTwilioStatus, twilioErrors } from "../twilio/call-resource.js"; import { serverErrors } from "./errors.js"; @@ -23,7 +29,7 @@ import { CallStore, type CallRecord } from "./call-store.js"; import { isSafeBwId, type BwClient } from "../bw/client.js"; import { checkReadiness } from "./readiness.js"; import { safeEqual } from "./safe-equal.js"; -import { captureTwiml } from "./capture.js"; +import { captureTwiml, captureStreamFrame } from "./capture.js"; export interface ServerConfig { accountSid: string; @@ -37,8 +43,16 @@ export interface ServerConfig { /** Basic-auth credentials Bandwidth presents on inbound webhooks (must match the app's CallbackCreds). */ webhookUser: string; webhookPassword: string; - /** Opt-in dir to persist each customer TwiML response for the later BXML Generator. Undefined → no capture. */ + /** Opt-in dir to persist each customer TwiML response for the later BXML Generator, and + * raw Bandwidth StartStream frames under /streams/. Undefined → no capture. */ captureDir?: string; + /** How long a Bandwidth StartStream WebSocket may sit without a "start" event before it is closed. Default 5000. */ + streamStartTimeoutMs?: number; + /** How long the bridge waits for the bot's WebSocket handshake before giving up and ending the + * Bandwidth stream. Bounds a bot that accepts TCP but never completes the upgrade. Default 10000. */ + streamBotConnectTimeoutMs?: number; + /** Passed to TwilioStreamBridge.playoutLatencyPadMs for every stream. Default 0. */ + streamPlayoutLatencyPadMs?: number; } export interface ServerDeps { @@ -110,6 +124,14 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc "Basic " + Buffer.from(`${config.accountSid}:${config.authToken}`).toString("base64"); const authOk = (req: FastifyRequest) => safeEqual(req.headers.authorization ?? "", expectedAuth); + // The WebSocket form of PUBLIC_BASE_URL: Bandwidth's StartStream destination + // must be ws(s)://, so https → wss and http → ws (local dev). + const publicWsBase = (() => { + const u = new URL(config.publicBaseUrl); + u.protocol = u.protocol === "http:" ? "ws:" : "wss:"; + return u.toString().replace(/\/$/, ""); + })(); + const rewriter = (base: string) => (url: string, kind: UrlKind) => { const absolute = new URL(url, base).toString(); // Recording-available events are async, fire-and-forget (no BXML continuation), @@ -117,9 +139,120 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc if (kind === "recordingStatus") { return `${config.publicBaseUrl}/bw/recording-status?cb=${encodeURIComponent(absolute)}`; } + // A Twilio is the customer's bot. Bandwidth speaks its own + // StartStream protocol, so the stream must come to us first; /bw/stream + // bridges it to the bot in Twilio's Media Streams protocol. + if (kind === "stream") { + return `${publicWsBase}/bw/stream?dest=${encodeURIComponent(absolute)}`; + } return `${config.publicBaseUrl}/bw/continue?next=${encodeURIComponent(absolute)}`; }; + // ── Media Streams: Bandwidth → /bw/stream → TwilioStreamBridge → bot ─────── + // Fastify has no WebSocket routing of its own, so the HTTP upgrade is taken + // off the underlying server and handed to ws. The upgrade must carry the same + // Basic-auth credentials as the /bw/* webhooks (the translator stamps them on + // StartStream as destinationUsername/destinationPassword), and `dest` goes + // through the egress guard like any customer-supplied URL. Once Bandwidth's + // "start" event arrives, a bridge is built per stream; caller audio is held + // until the bot's socket is open so the first words are not lost. + const streamServer = new WebSocketServer({ noServer: true }); + const bridges = new Set(); + + const rejectUpgrade = (socket: Duplex, status: number, reason: string, extraHeaders = ""): void => { + if (socket.destroyed) return; + socket.write(`HTTP/1.1 ${status} ${reason}\r\n${extraHeaders}Connection: close\r\nContent-Length: 0\r\n\r\n`); + socket.destroy(); + }; + + async function handleStreamUpgrade(req: IncomingMessage, socket: Duplex, head: Buffer): Promise { + // Node hands over the raw socket with no error listener; ws attaches one only + // inside handleUpgrade, and the egress check below can await a DNS lookup for + // seconds. A reset in that window would otherwise be an uncaught exception. + socket.on("error", () => socket.destroy()); + const url = new URL(req.url ?? "/", "http://placeholder.invalid"); + if (url.pathname !== "/bw/stream") return rejectUpgrade(socket, 404, "Not Found"); + if (!safeEqual(req.headers.authorization ?? "", expectedWebhookAuth)) { + return rejectUpgrade(socket, 401, "Unauthorized", "WWW-Authenticate: Basic\r\n"); + } + const dest = url.searchParams.get("dest"); + if (!dest) return rejectUpgrade(socket, 400, "Bad Request"); + try { + await assertPublicUrl(dest, { + allowPrivate: config.allowPrivateEgress, + allowHosts: config.egressAllowHosts, + schemes: ["ws:", "wss:"], + }); + } catch (err) { + app.log.error({ dest, err }, "stream dest blocked"); + return err instanceof EgressBlockedError + ? rejectUpgrade(socket, 400, "Bad Request") + : rejectUpgrade(socket, 500, "Internal Server Error"); + } + if (socket.destroyed) return; + + streamServer.handleUpgrade(req, socket, head, (ws) => { + // Per-connection key for capture; never derived from Bandwidth input. + const captureKey = randomUUID(); + const source = new BwWebSocketSource(ws, { + onFrame: config.captureDir + ? (raw) => { + try { + captureStreamFrame(config.captureDir!, captureKey, raw); + } catch (err) { + app.log.error({ err }, "stream capture failed"); + } + } + : undefined, + }); + let bridge: TwilioStreamBridge | undefined; + source + .waitForStart(config.streamStartTimeoutMs ?? 5000) + .then((start) => { + bridge = new TwilioStreamBridge({ + botUrl: dest, + callSid: toCallSid(start.callId), + accountSid: config.accountSid, + customParameters: customParametersFromBwStart(start.raw), + source, + playoutLatencyPadMs: config.streamPlayoutLatencyPadMs, + connectTimeoutMs: config.streamBotConnectTimeoutMs, + }); + const b = bridge; + bridges.add(b); + source.once("stop", () => bridges.delete(b)); + app.log.info({ callId: start.callId, streamId: start.streamId, dest }, "stream bridge started"); + return b.ready(); + }) + .then(() => source.release()) + .catch((err) => { + // No start, or the bot could not be reached: end the Bandwidth stream so + // the returns and the call's BXML moves on. + app.log.error({ dest, err }, "stream bridge setup failed"); + if (bridge) { + bridges.delete(bridge); + bridge.close(); + } + source.close(); + }); + }); + } + + app.server.on("upgrade", (req, socket, head) => { + void handleStreamUpgrade(req, socket, head); + }); + // preClose, not onClose: Fastify waits for the HTTP server's connections to + // drain before onClose runs, and an upgraded socket counts as one of them, so + // it must be torn down first or close() would wait for Bandwidth to hang up. + app.addHook("preClose", async () => { + for (const b of bridges) b.close(); + bridges.clear(); + // In noServer mode close() does not end established clients, and a socket + // that upgraded but has not sent "start" yet has no bridge to close it. + for (const client of streamServer.clients) client.terminate(); + streamServer.close(); + }); + function errorBxml(verbs: string[]): string { return bxmlDocument([ { diff --git a/src/server/capture.ts b/src/server/capture.ts index 01edfa5..b868230 100644 --- a/src/server/capture.ts +++ b/src/server/capture.ts @@ -1,7 +1,40 @@ -import { mkdirSync, writeFileSync, existsSync } from "node:fs"; +import { mkdirSync, writeFileSync, existsSync, appendFileSync } from "node:fs"; import { createHash } from "node:crypto"; import { join } from "node:path"; +/** Media frames kept per stream capture; start and stop are always written. */ +const MAX_CAPTURED_MEDIA_FRAMES = 10; +const capturedMedia = new Map(); + +/** + * Append one raw Bandwidth StartStream WebSocket frame to + * `/streams/.jsonl`, one JSON frame per line, so a live call yields + * checked-in fixtures for the bridge tests (see test/fixtures/bandwidth/). + * + * `streamKey` is a per-connection identifier chosen by the server (never + * Bandwidth-supplied), hashed like captureTwiml so nothing untrusted reaches the + * filesystem path. Only the first MAX_CAPTURED_MEDIA_FRAMES media frames are + * kept; a call is 50 frames a second and the shapes are identical. Same + * private-by-default modes and synchronous I/O as captureTwiml. + * + * Returns the path written, or undefined when the frame was skipped by the cap. + */ +export function captureStreamFrame(dir: string, streamKey: string, rawFrame: string): string | undefined { + const streamsDir = join(dir, "streams"); + const file = join(streamsDir, `${createHash("sha256").update(streamKey).digest("hex").slice(0, 16)}.jsonl`); + if (/"eventType"\s*:\s*"media"/.test(rawFrame)) { + const n = capturedMedia.get(file) ?? 0; + if (n >= MAX_CAPTURED_MEDIA_FRAMES) return undefined; + capturedMedia.set(file, n + 1); + } else if (/"eventType"\s*:\s*"stop"/.test(rawFrame)) { + // The stream is over; forget its tally so the map does not grow per call. + capturedMedia.delete(file); + } + mkdirSync(streamsDir, { recursive: true, mode: 0o700 }); + appendFileSync(file, rawFrame.replace(/\r?\n/g, " ") + "\n", { mode: 0o600 }); + return file; +} + /** * Persist a raw TwiML response the translator fetched from the customer app, so a * later `npm run bxml-generator` over the capture dir can turn the paths a test call diff --git a/src/server/index.ts b/src/server/index.ts index 8afdea7..2f7ea66 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -9,6 +9,14 @@ function env(name: string): string { const bwEnv = process.env.BW_ENVIRONMENT === "test" ? "test" : "prod"; +/** Optional millisecond setting; unset or non-numeric leaves the code default in place. */ +function optionalMs(name: string): number | undefined { + const v = process.env[name]; + if (!v) return undefined; + const n = Number(v); + return Number.isFinite(n) && n >= 0 ? n : undefined; +} + const app = buildApp( { accountSid: env("TRANSLATOR_ACCOUNT_SID"), @@ -20,6 +28,9 @@ const app = buildApp( allowPrivateEgress: process.env.EGRESS_ALLOW_PRIVATE === "1", egressAllowHosts: process.env.EGRESS_ALLOW_HOSTS?.split(",").map((s) => s.trim()).filter(Boolean), captureDir: process.env.TRANSLATOR_CAPTURE_DIR, + streamPlayoutLatencyPadMs: optionalMs("STREAM_PLAYOUT_LATENCY_PAD_MS"), + streamStartTimeoutMs: optionalMs("STREAM_START_TIMEOUT_MS"), + streamBotConnectTimeoutMs: optionalMs("STREAM_BOT_CONNECT_TIMEOUT_MS"), }, { fetchImpl: fetch, diff --git a/src/streams/bridge.ts b/src/streams/bridge.ts index 2b48a82..f368ecd 100644 --- a/src/streams/bridge.ts +++ b/src/streams/bridge.ts @@ -50,8 +50,14 @@ export interface BridgeOpts { * return slightly early in the same direction as the original bug. Default 0 * until real measurements from VAPI-3991 give a value worth setting. */ playoutLatencyPadMs?: number; + /** Bot WebSocket handshake timeout in ms; ready() rejects when it elapses. Default 10000. + * Without it a bot that accepts TCP but never answers the upgrade holds the stream + * open until the OS gives up, which can be a minute or more of silence for the caller. */ + connectTimeoutMs?: number; } +const DEFAULT_CONNECT_TIMEOUT_MS = 10_000; + /** * Map Bandwidth's StartStream WebSocket "start" event to Twilio `customParameters`. * @@ -100,7 +106,7 @@ export class TwilioStreamBridge { constructor(private opts: BridgeOpts) { this.streamSid = "MZ" + randomBytes(16).toString("hex"); - this.ws = new WebSocket(opts.botUrl); + this.ws = new WebSocket(opts.botUrl, { handshakeTimeout: opts.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS }); this.readyPromise = new Promise((resolve, reject) => { this.ws.on("open", () => { // 1. connected — protocol handshake (no sequenceNumber per Twilio spec) @@ -168,6 +174,14 @@ export class TwilioStreamBridge { this.ws.close(); }); + // Bot hung up its side. On Twilio that ends and TwiML resumes + // after it; here, closing the Bandwidth socket ends the StartStream so the + // translator's returns and BXML resumes likewise. + this.ws.on("close", () => { + this.dropPendingMarks(); + this.opts.source.close(); + }); + // Bot → bridge inbound message handler. The bot only ever sends // media / mark / clear back to us (per Twilio's protocol). this.ws.on("message", (data) => { diff --git a/src/streams/bw-source.ts b/src/streams/bw-source.ts new file mode 100644 index 0000000..0b95a6c --- /dev/null +++ b/src/streams/bw-source.ts @@ -0,0 +1,216 @@ +import { EventEmitter } from "node:events"; +import { WebSocket } from "ws"; +import type { BwStreamSource } from "./bridge.js"; + +/** + * The Bandwidth side of the Media Streams bridge: a BwStreamSource over the + * WebSocket Bandwidth opens to a (VAPI-3991). + * + * Protocol (StartStream docs, confirmed on real calls 2026-09-17), all JSON: + * + * Bandwidth → us + * { eventType: "start", metadata: { accountId, callId, streamId, streamName, + * tracks: [{ name, mediaFormat: { encoding: "PCMU", sampleRate: 8000 } }] }, + * streamParams?: { name: value } } + * { eventType: "media", track: "inbound" | "outbound", payload: , + * sequenceNumber: "1" } + * { eventType: "stop", metadata: { ...same shape as start } } + * + * us → Bandwidth (bidirectional streams only) + * { eventType: "playAudio", media: { contentType: "audio/pcmu", payload } } + * { eventType: "clear" } + * + * Bandwidth does not deliver DTMF over this socket (it arrives via BXML Gather + * webhooks), so this source never emits "dtmf". + * + * Only the "inbound" track (the caller) is relayed: the bridge presents a single + * inbound track to the bot, matching what the translator emits (tracks="inbound"). + * Outbound-track frames, if a BXML author asked for them, are ignored. + * + * Media that arrives before release() is buffered (bounded), so the server can + * wait for the bot's WebSocket to open before the first caller audio flows and + * nothing said in the first few hundred milliseconds is lost. + */ + +export interface BwStreamTrack { + name: string; + mediaFormat?: { encoding?: string; sampleRate?: number }; +} + +export interface BwStreamStart { + accountId: string; + callId: string; + streamId: string; + streamName: string; + tracks: BwStreamTrack[]; + /** values echoed by Bandwidth, as a flat map. */ + streamParams: Record; + /** The start event exactly as received (input to customParametersFromBwStart). */ + raw: unknown; +} + +export interface BwWebSocketSourceOpts { + /** Receives every inbound frame's raw text before parsing; used for fixture capture. */ + onFrame?: (raw: string) => void; +} + +/** Frames of caller audio held before release(): 500 × 20 ms = 10 s, oldest dropped first. */ +const MAX_BUFFERED_FRAMES = 500; + +function str(v: unknown): string { + return typeof v === "string" ? v : v === undefined || v === null ? "" : String(v); +} + +function parseStart(msg: Record): BwStreamStart { + const md = (msg.metadata ?? {}) as Record; + const tracks: BwStreamTrack[] = Array.isArray(md.tracks) + ? (md.tracks as unknown[]).flatMap((t) => { + if (!t || typeof t !== "object") return []; + const tr = t as Record; + const mf = (tr.mediaFormat ?? undefined) as Record | undefined; + return [ + { + name: str(tr.name), + ...(mf ? { mediaFormat: { encoding: str(mf.encoding), sampleRate: Number(mf.sampleRate) || undefined } } : {}), + }, + ]; + }) + : []; + const streamParams: Record = Object.create(null); + const sp = msg.streamParams; + if (sp && typeof sp === "object" && !Array.isArray(sp)) { + for (const [k, v] of Object.entries(sp as Record)) { + if (typeof v === "string") streamParams[k] = v; + else if (typeof v === "number" || typeof v === "boolean") streamParams[k] = String(v); + } + } + return { + accountId: str(md.accountId), + callId: str(md.callId), + streamId: str(md.streamId), + streamName: str(md.streamName), + tracks, + streamParams, + raw: msg, + }; +} + +export class BwWebSocketSource extends EventEmitter implements BwStreamSource { + /** Set once Bandwidth's start event has been parsed. */ + start: BwStreamStart | undefined; + private released = false; + private buffered: string[] = []; + private stopped = false; + + constructor( + private ws: WebSocket, + private opts: BwWebSocketSourceOpts = {}, + ) { + super(); + ws.on("message", (data) => this.onMessage(String(data))); + // "close" always follows "error" on ws; nothing to do here except keep an + // unhandled error from crashing the process. + ws.on("error", () => {}); + ws.on("close", () => this.emitStop()); + } + + /** + * Resolve with the parsed start event, or reject if the socket closes or + * `timeoutMs` passes first. Bandwidth sends start as its first frame, so a + * timeout means whatever connected is not a Bandwidth stream. + */ + waitForStart(timeoutMs: number): Promise { + if (this.start) return Promise.resolve(this.start); + if (this.stopped) return Promise.reject(new Error("stream closed before start")); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup(); + reject(new Error(`no start event within ${timeoutMs} ms`)); + }, timeoutMs); + timer.unref?.(); // never the only thing keeping the process (or a test) alive + const onStart = (s: BwStreamStart) => { + cleanup(); + resolve(s); + }; + const onStop = () => { + cleanup(); + reject(new Error("stream closed before start")); + }; + const cleanup = () => { + clearTimeout(timer); + this.off("start", onStart); + this.off("stop", onStop); + }; + this.once("start", onStart); + this.once("stop", onStop); + }); + } + + /** Start emitting "media"; frames received so far are emitted first, in order. */ + release(): void { + if (this.released) return; + this.released = true; + const pending = this.buffered; + this.buffered = []; + for (const p of pending) this.emit("media", p); + } + + /** Frames of caller audio currently held back (0 once released). */ + bufferedFrames(): number { + return this.buffered.length; + } + + sendMedia(payloadB64: string): void { + this.sendJson({ eventType: "playAudio", media: { contentType: "audio/pcmu", payload: payloadB64 } }); + } + + flush(): void { + this.sendJson({ eventType: "clear" }); + } + + close(): void { + if (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING) this.ws.close(); + } + + private sendJson(obj: unknown): void { + if (this.ws.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(obj)); + } + + private onMessage(raw: string): void { + this.opts.onFrame?.(raw); + let msg: unknown; + try { + msg = JSON.parse(raw); + } catch { + return; // not one of ours; ignore rather than tear the stream down + } + if (!msg || typeof msg !== "object") return; + const m = msg as Record; + switch (m.eventType) { + case "start": + this.start = parseStart(m); + this.emit("start", this.start); + break; + case "media": { + if (typeof m.payload !== "string") return; + if (m.track !== undefined && m.track !== "inbound") return; + if (this.released) { + this.emit("media", m.payload); + } else { + if (this.buffered.length >= MAX_BUFFERED_FRAMES) this.buffered.shift(); + this.buffered.push(m.payload); + } + break; + } + case "stop": + this.emitStop(); + break; + } + } + + private emitStop(): void { + if (this.stopped) return; + this.stopped = true; + this.emit("stop"); + } +} diff --git a/src/translator/translate.ts b/src/translator/translate.ts index 5c10ec6..3ad9e3f 100644 --- a/src/translator/translate.ts +++ b/src/translator/translate.ts @@ -178,13 +178,24 @@ const CALLBACK_URL_ATTRS = [ /** Walks the built element tree and stamps username/password onto any element * carrying a rewritten translator callback URL, so Bandwidth Basic-auths the - * continuation request instead of hitting it unauthenticated. */ + * continuation request instead of hitting it unauthenticated. A StartStream's + * destination is the translator's own /bw/stream WebSocket after rewriting, so + * it gets the same credentials as destinationUsername/destinationPassword, + * which Bandwidth presents as the Authorization header on the upgrade. */ function stampCallbackAuth(els: XmlEl[], auth: { username: string; password: string }): void { for (const el of els) { if (el.attrs && CALLBACK_URL_ATTRS.some((a) => el.attrs![a] !== undefined)) { el.attrs.username = auth.username; el.attrs.password = auth.password; } + // INVARIANT: callbackAuth is only passed together with the server's rewriter, + // which points every "stream" URL at the translator's own /bw/stream. These + // are the translator's webhook credentials; a rewriter that ever left the + // customer's bot URL in place would hand them to a third party. + if (el.name === "StartStream" && el.attrs?.destination !== undefined) { + el.attrs.destinationUsername = auth.username; + el.attrs.destinationPassword = auth.password; + } if (el.children) { const childEls = el.children.filter( (c): c is XmlEl => typeof c !== "string" && !("raw" in c), diff --git a/src/twilio/egress-guard.ts b/src/twilio/egress-guard.ts index e028723..3bf1cd6 100644 --- a/src/twilio/egress-guard.ts +++ b/src/twilio/egress-guard.ts @@ -40,7 +40,13 @@ export function isBlockedAddress(ip: string): boolean { export async function assertPublicUrl( rawUrl: string, - opts: { allowPrivate?: boolean; allowHosts?: string[]; lookup?: (host: string) => Promise } = {}, + opts: { + allowPrivate?: boolean; + allowHosts?: string[]; + lookup?: (host: string) => Promise; + /** Accepted URL schemes (with colon). Default http/https; the stream route passes ws/wss. */ + schemes?: string[]; + } = {}, ): Promise { let url: URL; try { @@ -48,8 +54,8 @@ export async function assertPublicUrl( } catch { throw new EgressBlockedError(`Malformed URL: ${rawUrl}`); } - if (url.protocol !== "http:" && url.protocol !== "https:") - throw new EgressBlockedError(`Disallowed scheme: ${url.protocol}`); + const schemes = opts.schemes ?? ["http:", "https:"]; + if (!schemes.includes(url.protocol)) throw new EgressBlockedError(`Disallowed scheme: ${url.protocol}`); // Opt-in host allowlist ("full remediation"): default-deny by host. Listed // hosts are explicitly trusted, so they bypass the range denylist below. diff --git a/test/fixtures/bandwidth/stream-frames.json b/test/fixtures/bandwidth/stream-frames.json new file mode 100644 index 0000000..8c4f135 --- /dev/null +++ b/test/fixtures/bandwidth/stream-frames.json @@ -0,0 +1,64 @@ +{ + "_meta": { + "captured": "2026-09-17", + "source": "start.metadata: live Bandwidth on a real call, recorded by the local repro app. Field names, nesting, and the PCMU/8000 track format are exactly as received; ids redacted.", + "note": "streamParams follows the StartStream docs example; media and stop follow the docs' documented shapes; the media payload is 160 bytes (20 ms) of synthetic PCMU silence, not recorded audio. On 2026-09-25 a live inbound call ran end to end through /bw/stream with frame capture on: the real start, media, and stop frames matched these shapes field for field (Bandwidth orders eventType last, which the parser does not depend on), streamParams arrived as the flat map modelled here, and media frames were 160-byte inbound PCMU with sequenceNumber counting from \"1\". Those recorded frames are deliberately not checked in: they came from a personal line and this repo is public. To refresh from a shared test account, run a call with TRANSLATOR_CAPTURE_DIR set and copy /streams/*.jsonl.", + "redacted": [ + "accountId", + "callId", + "streamId" + ] + }, + "start": { + "eventType": "start", + "metadata": { + "accountId": "99XXXXX", + "callId": "c-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamId": "s-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamName": "connect-stream-1", + "tracks": [ + { + "name": "inbound", + "mediaFormat": { + "encoding": "PCMU", + "sampleRate": 8000 + } + } + ] + }, + "streamParams": { + "callSid": "CA123", + "tenant": "acme" + } + }, + "media": { + "eventType": "media", + "track": "inbound", + "payload": "/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////w==", + "sequenceNumber": "1" + }, + "mediaOutbound": { + "eventType": "media", + "track": "outbound", + "payload": "/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////w==", + "sequenceNumber": "1" + }, + "stop": { + "eventType": "stop", + "metadata": { + "accountId": "99XXXXX", + "callId": "c-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamId": "s-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamName": "connect-stream-1", + "tracks": [ + { + "name": "inbound", + "mediaFormat": { + "encoding": "PCMU", + "sampleRate": 8000 + } + } + ] + } + } +} diff --git a/test/server-stream.test.ts b/test/server-stream.test.ts new file mode 100644 index 0000000..34a4c72 --- /dev/null +++ b/test/server-stream.test.ts @@ -0,0 +1,358 @@ +/** + * /bw/stream: the translator's StartStream destination (VAPI-3991). + * + * Bandwidth (played by a ws client) connects to the running Fastify server with + * the fixture frames; a local WebSocketServer plays the customer's Twilio bot. + * Everything runs on 127.0.0.1 with allowPrivateEgress so ports are ephemeral. + */ +import { describe, it, expect, vi } from "vitest"; +import { WebSocketServer, WebSocket } from "ws"; +import { createServer, type AddressInfo } from "node:net"; +import { readFileSync, mkdtempSync, rmSync, readdirSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { buildApp } from "../src/server/app.js"; +import { toCallSid } from "../src/twilio/call-sid.js"; + +const frames = JSON.parse( + readFileSync(new URL("./fixtures/bandwidth/stream-frames.json", import.meta.url), "utf8"), +); +const webhookAuth = "Basic " + Buffer.from("u:p").toString("base64"); + +function config(extra: Record = {}) { + return { + accountSid: "AC123", + authToken: "tok", + publicBaseUrl: "https://translator.test", + voiceUrl: "https://customer.test/voice", + allowPrivateEgress: true, + webhookUser: "u", + webhookPassword: "p", + streamStartTimeoutMs: 300, + ...extra, + }; +} + +function makeApp(cfg = config(), twimlByUrl: Record = {}) { + const fetchImpl = vi.fn(async (url: any) => { + const twiml = twimlByUrl[String(url)]; + return twiml ? new Response(twiml, { status: 200 }) : new Response("not found", { status: 404 }); + }) as unknown as typeof fetch; + const bwClient = { + createCall: vi.fn(), modifyCall: vi.fn(), getCall: vi.fn(), listRecordings: vi.fn(), + getRecording: vi.fn(), getRecordingMedia: vi.fn(), updateRecording: vi.fn(), + }; + return buildApp(cfg, { fetchImpl, bwClient }); +} + +async function listen(app: ReturnType): Promise { + await app.listen({ port: 0, host: "127.0.0.1" }); + return (app.server.address() as AddressInfo).port; +} + +/** The customer's bot, speaking Twilio Media Streams. */ +async function fakeBot() { + const wss = new WebSocketServer({ port: 0, host: "127.0.0.1" }); + await new Promise((r) => wss.once("listening", r)); + const port = (wss.address() as AddressInfo).port; + const messages: any[] = []; + const socket = new Promise((res) => + wss.on("connection", (ws) => { + ws.on("message", (d) => messages.push(JSON.parse(String(d)))); + res(ws); + }), + ); + return { url: `ws://127.0.0.1:${port}/bot`, messages, socket, close: () => wss.close() }; +} + +/** Connect as Bandwidth would: the rewritten destination plus destination creds. */ +function bandwidthClient(port: number, dest: string, headers: Record = { authorization: webhookAuth }) { + const ws = new WebSocket(`ws://127.0.0.1:${port}/bw/stream?dest=${encodeURIComponent(dest)}`, { headers }); + const received: any[] = []; + ws.on("message", (d) => received.push(JSON.parse(String(d)))); + return { ws, received, send: (obj: unknown) => ws.send(JSON.stringify(obj)) }; +} + +/** 101 on a completed upgrade, otherwise the HTTP status the server rejected with. */ +function upgradeStatus(ws: WebSocket): Promise { + return new Promise((resolve, reject) => { + ws.once("open", () => resolve(101)); + ws.once("unexpected-response", (_req, res) => { + res.resume(); + resolve(res.statusCode ?? 0); + }); + ws.once("error", reject); + }); +} + +async function waitFor(check: () => boolean, ms = 1000): Promise { + const deadline = Date.now() + ms; + while (!check()) { + if (Date.now() > deadline) throw new Error("waitFor timeout"); + await new Promise((r) => setTimeout(r, 10)); + } +} + +describe("/bw/stream end to end", () => { + it("bridges a Bandwidth stream to the bot in Twilio's protocol, both directions, then stops", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + expect(await upgradeStatus(bw.ws)).toBe(101); + + // Bandwidth → bot: start becomes connected + start with our ids and customParameters. + bw.send(frames.start); + const botWs = await bot.socket; + await waitFor(() => bot.messages.length >= 2); + expect(bot.messages[0]).toMatchObject({ event: "connected", protocol: "Call" }); + expect(bot.messages[1].event).toBe("start"); + expect(bot.messages[1].start.callSid).toBe(toCallSid(frames.start.metadata.callId)); + expect(bot.messages[1].start.accountSid).toBe("AC123"); + expect(bot.messages[1].start.customParameters).toEqual({ callSid: "CA123", tenant: "acme" }); + expect(bot.messages[1].start.mediaFormat).toEqual({ encoding: "audio/x-mulaw", sampleRate: 8000, channels: 1 }); + + // Caller audio reaches the bot as a Twilio media frame. + bw.send(frames.media); + await waitFor(() => bot.messages.some((m) => m.event === "media")); + expect(bot.messages.find((m) => m.event === "media").media).toMatchObject({ + track: "inbound", + payload: frames.media.payload, + }); + + // Bot → Bandwidth: media becomes playAudio, clear becomes clear. + const streamSid = bot.messages[1].streamSid; + botWs.send(JSON.stringify({ event: "media", streamSid, media: { payload: "QUJD" } })); + botWs.send(JSON.stringify({ event: "clear", streamSid })); + await waitFor(() => bw.received.length >= 2); + expect(bw.received[0]).toEqual({ eventType: "playAudio", media: { contentType: "audio/pcmu", payload: "QUJD" } }); + expect(bw.received[1]).toEqual({ eventType: "clear" }); + + // Bandwidth ends the stream: the bot gets stop and its socket is closed. + const botClosed = new Promise((r) => botWs.once("close", r)); + bw.send(frames.stop); + await botClosed; + expect(bot.messages.at(-1)).toMatchObject({ event: "stop", stop: { accountSid: "AC123" } }); + + bw.ws.close(); + bot.close(); + await app.close(); + }); + + it("holds caller audio sent before the bot's socket is open and delivers it in order", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + + bw.send(frames.start); + for (const p of ["AAAA", "BBBB", "CCCC"]) bw.send({ ...frames.media, payload: p }); + + await waitFor(() => bot.messages.filter((m) => m.event === "media").length === 3); + expect(bot.messages.filter((m) => m.event === "media").map((m) => m.media.payload)).toEqual(["AAAA", "BBBB", "CCCC"]); + // And they arrive after connected/start, never before. + expect(bot.messages.findIndex((m) => m.event === "media")).toBeGreaterThan(1); + + bw.ws.close(); + bot.close(); + await app.close(); + }); + + it("ends the Bandwidth stream when the bot closes its socket", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + bw.send(frames.start); + const botWs = await bot.socket; + await waitFor(() => bot.messages.length >= 2); + + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + botWs.close(); + await bwClosed; + + bot.close(); + await app.close(); + }); + + it("ends the Bandwidth stream when the bot cannot be reached", async () => { + const app = makeApp(); + const port = await listen(app); + // Nothing listens on this port; the bridge's connect fails. + const bw = bandwidthClient(port, "ws://127.0.0.1:1/bot"); + await upgradeStatus(bw.ws); + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + bw.send(frames.start); + await bwClosed; + await app.close(); + }); + + it("ends the Bandwidth stream when the bot accepts TCP but never completes the WebSocket handshake", async () => { + // A black-hole bot: accepts the connection and says nothing. Without the + // handshake timeout the stream would hang until the OS gave up. + const blackHole = createServer((sock) => sock.on("error", () => {})); + await new Promise((r) => blackHole.listen(0, "127.0.0.1", r)); + const holePort = (blackHole.address() as AddressInfo).port; + + const app = makeApp(config({ streamBotConnectTimeoutMs: 300 })); + const port = await listen(app); + const bw = bandwidthClient(port, `ws://127.0.0.1:${holePort}/bot`); + await upgradeStatus(bw.ws); + const sentAt = Date.now(); + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + bw.send(frames.start); + await bwClosed; + // Long enough to be the timeout, not an immediate refusal; generous upper bound for slow CI. + expect(Date.now() - sentAt).toBeGreaterThanOrEqual(250); + expect(Date.now() - sentAt).toBeLessThan(5000); + + blackHole.close(); + await app.close(); + }); + + it("closes a socket that sends no start event within the timeout", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + const openedAt = Date.now(); + await new Promise((r) => bw.ws.once("close", r)); + expect(Date.now() - openedAt).toBeGreaterThanOrEqual(250); + // Upper bound is deliberately loose: it only guards against "never closes". + expect(Date.now() - openedAt).toBeLessThan(5000); + bot.close(); + await app.close(); + }); + + it("app.close() terminates a stream that upgraded but has not sent start yet", async () => { + const app = makeApp(config({ streamStartTimeoutMs: 60_000 })); + const port = await listen(app); + const bw = bandwidthClient(port, "ws://127.0.0.1:9/bot"); + await upgradeStatus(bw.ws); + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + const closingAt = Date.now(); + await app.close(); + await bwClosed; + expect(Date.now() - closingAt).toBeLessThan(2000); + }); +}); + +describe("/bw/stream upgrade gate", () => { + it("rejects an upgrade without the webhook credentials (401)", async () => { + const app = makeApp(); + const port = await listen(app); + const bw = bandwidthClient(port, "ws://127.0.0.1:9/bot", {}); + expect(await upgradeStatus(bw.ws)).toBe(401); + const wrong = bandwidthClient(port, "ws://127.0.0.1:9/bot", { + authorization: "Basic " + Buffer.from("u:wrong").toString("base64"), + }); + expect(await upgradeStatus(wrong.ws)).toBe(401); + await app.close(); + }); + + it("rejects a missing dest (400)", async () => { + const app = makeApp(); + const port = await listen(app); + const ws = new WebSocket(`ws://127.0.0.1:${port}/bw/stream`, { headers: { authorization: webhookAuth } }); + expect(await upgradeStatus(ws)).toBe(400); + await app.close(); + }); + + it("rejects a dest that is not a WebSocket URL (400)", async () => { + const app = makeApp(); + const port = await listen(app); + const bw = bandwidthClient(port, "https://bot.example/audio"); + expect(await upgradeStatus(bw.ws)).toBe(400); + await app.close(); + }); + + it("rejects a private-network dest when private egress is not allowed (400)", async () => { + const app = makeApp(config({ allowPrivateEgress: false })); + const port = await listen(app); + const bw = bandwidthClient(port, "ws://127.0.0.1:9/bot"); + expect(await upgradeStatus(bw.ws)).toBe(400); + await app.close(); + }); + + it("rejects an upgrade on any other path (404)", async () => { + const app = makeApp(); + const port = await listen(app); + const ws = new WebSocket(`ws://127.0.0.1:${port}/bw/other?dest=ws%3A%2F%2Fx`, { headers: { authorization: webhookAuth } }); + expect(await upgradeStatus(ws)).toBe(404); + await app.close(); + }); +}); + +describe("translator side of the stream route", () => { + it("rewrites a Connect/Stream destination to wss:///bw/stream?dest= and stamps destination creds", async () => { + const app = makeApp(config(), { + "https://customer.test/voice": ``, + }); + const res = await app.inject({ + method: "POST", + url: "/bw/initiate", + headers: { authorization: webhookAuth }, + payload: { eventType: "initiate", callId: "c-stream-1", from: "+1", to: "+2", direction: "inbound" }, + }); + expect(res.statusCode).toBe(200); + expect(res.body).toContain(`destination="wss://translator.test/bw/stream?dest=wss%3A%2F%2Fbot.example%2Faudio"`); + expect(res.body).toContain(`destinationUsername="u"`); + expect(res.body).toContain(`destinationPassword="p"`); + expect(res.body).toContain(``); + expect(res.body).toContain(``); + }); + + it("uses ws:// when PUBLIC_BASE_URL is plain http (local dev)", async () => { + const app = makeApp(config({ publicBaseUrl: "http://localhost:3000" }), { + "https://customer.test/voice": ``, + }); + const res = await app.inject({ + method: "POST", + url: "/bw/initiate", + headers: { authorization: webhookAuth }, + payload: { eventType: "initiate", callId: "c-stream-2", from: "+1", to: "+2", direction: "inbound" }, + }); + expect(res.body).toContain(`destination="ws://localhost:3000/bw/stream?dest=`); + }); +}); + +describe("stream frame capture", () => { + it("writes start, the first 10 media frames, and stop to /streams/*.jsonl", async () => { + const dir = mkdtempSync(join(tmpdir(), "stream-capture-")); + try { + const app = makeApp(config({ captureDir: dir })); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + bw.send(frames.start); + const botWs = await bot.socket; + // 12 frames sent; the cap keeps 10 so a capture stays small. + for (let i = 1; i <= 12; i++) bw.send({ ...frames.media, sequenceNumber: String(i) }); + await waitFor(() => bot.messages.filter((m) => m.event === "media").length === 12); + const botClosed = new Promise((r) => botWs.once("close", r)); + bw.send(frames.stop); + await botClosed; + bw.ws.close(); + bot.close(); + await app.close(); + + const files = readdirSync(join(dir, "streams")); + expect(files).toHaveLength(1); + expect(files[0]).toMatch(/^[0-9a-f]{16}\.jsonl$/); + const lines = readFileSync(join(dir, "streams", files[0]), "utf8").trim().split("\n").map((l) => JSON.parse(l)); + expect(lines[0].eventType).toBe("start"); + expect(lines[0].metadata.streamId).toBe(frames.start.metadata.streamId); + expect(lines.filter((l) => l.eventType === "media").map((l) => l.sequenceNumber)).toEqual( + ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"], + ); + expect(lines.at(-1).eventType).toBe("stop"); + expect(lines).toHaveLength(12); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/streams-bw-source.test.ts b/test/streams-bw-source.test.ts new file mode 100644 index 0000000..f18106f --- /dev/null +++ b/test/streams-bw-source.test.ts @@ -0,0 +1,232 @@ +/** + * BwWebSocketSource: the Bandwidth side of the Media Streams bridge (VAPI-3991). + * + * Pattern: a local WebSocketServer stands in for the translator's /bw/stream + * endpoint and a ws client plays "Bandwidth", sending the frames in + * test/fixtures/bandwidth/stream-frames.json. + */ +import { describe, it, expect } from "vitest"; +import { WebSocketServer, WebSocket } from "ws"; +import { readFileSync } from "node:fs"; +import type { AddressInfo } from "node:net"; +import { BwWebSocketSource, type BwWebSocketSourceOpts } from "../src/streams/bw-source.js"; +import { customParametersFromBwStart } from "../src/streams/bridge.js"; + +const frames = JSON.parse( + readFileSync(new URL("./fixtures/bandwidth/stream-frames.json", import.meta.url), "utf8"), +); + +async function pair(opts?: BwWebSocketSourceOpts) { + const wss = new WebSocketServer({ port: 0, host: "127.0.0.1" }); + await new Promise((r) => wss.once("listening", r)); + const port = (wss.address() as AddressInfo).port; + const sourceP = new Promise((res) => + wss.once("connection", (ws) => res(new BwWebSocketSource(ws, opts))), + ); + const bandwidth = new WebSocket(`ws://127.0.0.1:${port}/bw/stream`); + const received: any[] = []; + bandwidth.on("message", (d) => received.push(JSON.parse(String(d)))); + await new Promise((r) => bandwidth.once("open", r)); + const source = await sourceP; + const send = (obj: unknown) => bandwidth.send(typeof obj === "string" ? obj : JSON.stringify(obj)); + return { + source, + bandwidth, + received, + send, + close: () => { + bandwidth.close(); + wss.close(); + }, + }; +} + +async function waitFor(check: () => boolean, ms = 500): Promise { + const deadline = Date.now() + ms; + while (!check()) { + if (Date.now() > deadline) throw new Error("waitFor timeout"); + await new Promise((r) => setTimeout(r, 10)); + } +} + +describe("BwWebSocketSource: start", () => { + it("parses the recorded start event: ids, tracks, and streamParams", async () => { + const t = await pair(); + t.send(frames.start); + const start = await t.source.waitForStart(1000); + t.close(); + + expect(start.accountId).toBe(frames.start.metadata.accountId); + expect(start.callId).toBe(frames.start.metadata.callId); + expect(start.streamId).toBe(frames.start.metadata.streamId); + expect(start.streamName).toBe("connect-stream-1"); + expect(start.tracks).toEqual([{ name: "inbound", mediaFormat: { encoding: "PCMU", sampleRate: 8000 } }]); + expect(start.streamParams).toEqual({ callSid: "CA123", tenant: "acme" }); + // The raw event feeds the bridge's customParameters mapper unchanged. + expect(customParametersFromBwStart(start.raw)).toEqual({ callSid: "CA123", tenant: "acme" }); + expect(t.source.start).toBe(start); + }); + + it("tolerates a start event with no metadata or streamParams", async () => { + const t = await pair(); + t.send({ eventType: "start" }); + const start = await t.source.waitForStart(1000); + t.close(); + expect(start).toMatchObject({ accountId: "", callId: "", streamId: "", streamName: "", tracks: [] }); + expect(start.streamParams).toEqual({}); + }); + + it("waitForStart resolves immediately if start already arrived", async () => { + const t = await pair(); + t.send(frames.start); + await waitFor(() => t.source.start !== undefined); + const start = await t.source.waitForStart(1); + t.close(); + expect(start.callId).toBe(frames.start.metadata.callId); + }); + + it("waitForStart rejects when no start arrives in time", async () => { + const t = await pair(); + await expect(t.source.waitForStart(50)).rejects.toThrow(/no start event within 50 ms/); + t.close(); + }); + + it("waitForStart rejects when Bandwidth closes first", async () => { + const t = await pair(); + const p = t.source.waitForStart(1000); + t.bandwidth.close(); + await expect(p).rejects.toThrow(/closed before start/); + t.close(); + }); +}); + +describe("BwWebSocketSource: media (Bandwidth → bot)", () => { + it("buffers inbound frames until release(), then emits them in order and streams the rest live", async () => { + const t = await pair(); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + + t.send({ ...frames.media, payload: "AAAA", sequenceNumber: "1" }); + t.send({ ...frames.media, payload: "BBBB", sequenceNumber: "2" }); + t.send({ ...frames.media, payload: "CCCC", sequenceNumber: "3" }); + await waitFor(() => t.source.bufferedFrames() === 3); + expect(got).toEqual([]); + + t.source.release(); + expect(got).toEqual(["AAAA", "BBBB", "CCCC"]); + expect(t.source.bufferedFrames()).toBe(0); + + t.send({ ...frames.media, payload: "DDDD", sequenceNumber: "4" }); + await waitFor(() => got.length === 4); + t.close(); + expect(got[3]).toBe("DDDD"); + }); + + it("relays the recorded media frame's payload verbatim", async () => { + const t = await pair(); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + t.source.release(); + t.send(frames.media); + await waitFor(() => got.length === 1); + t.close(); + expect(got[0]).toBe(frames.media.payload); + expect(Buffer.from(got[0], "base64")).toHaveLength(160); // 20 ms of PCMU + }); + + it("ignores outbound-track frames and frames without a string payload", async () => { + const t = await pair(); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + t.source.release(); + t.send(frames.mediaOutbound); + t.send({ eventType: "media", track: "inbound", payload: 123 }); + t.send({ eventType: "media", track: "inbound" }); + t.send({ ...frames.media, payload: "ZZZZ" }); + await waitFor(() => got.length === 1); + await new Promise((r) => setTimeout(r, 30)); + t.close(); + expect(got).toEqual(["ZZZZ"]); + }); + + it("drops the oldest buffered frames beyond the cap (10 s of audio)", async () => { + const t = await pair(); + for (let i = 0; i < 505; i++) t.send({ ...frames.media, payload: `f${i}`, sequenceNumber: String(i + 1) }); + await waitFor(() => t.source.bufferedFrames() === 500, 2000); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + t.source.release(); + t.close(); + expect(got).toHaveLength(500); + expect(got[0]).toBe("f5"); + expect(got[499]).toBe("f504"); + }); + + it("ignores non-JSON frames and unknown event types", async () => { + const t = await pair(); + t.send("not json at all"); + t.send({ eventType: "somethingNew", data: 1 }); + t.send(frames.start); + const start = await t.source.waitForStart(1000); + t.close(); + expect(start.streamName).toBe("connect-stream-1"); + }); +}); + +describe("BwWebSocketSource: stop", () => { + it("emits stop exactly once for a stop event followed by the socket closing", async () => { + const t = await pair(); + let stops = 0; + t.source.on("stop", () => stops++); + t.send(frames.stop); + await waitFor(() => stops === 1); + t.bandwidth.close(); + await new Promise((r) => setTimeout(r, 50)); + t.close(); + expect(stops).toBe(1); + }); + + it("emits stop when the socket closes without a stop event", async () => { + const t = await pair(); + let stops = 0; + t.source.on("stop", () => stops++); + t.bandwidth.close(); + await waitFor(() => stops === 1); + t.close(); + expect(stops).toBe(1); + }); +}); + +describe("BwWebSocketSource: bot → Bandwidth", () => { + it("sendMedia becomes playAudio audio/pcmu and flush becomes clear", async () => { + const t = await pair(); + t.source.sendMedia("QUJD"); + t.source.flush(); + await waitFor(() => t.received.length >= 2); + t.close(); + expect(t.received[0]).toEqual({ eventType: "playAudio", media: { contentType: "audio/pcmu", payload: "QUJD" } }); + expect(t.received[1]).toEqual({ eventType: "clear" }); + }); + + it("close() closes Bandwidth's socket and is safe to call twice", async () => { + const t = await pair(); + const closed = new Promise((r) => t.bandwidth.once("close", r)); + t.source.close(); + t.source.close(); + await closed; + t.close(); + }); + + it("onFrame receives every raw frame before parsing", async () => { + const raws: string[] = []; + const t = await pair({ onFrame: (raw) => raws.push(raw) }); + t.send(frames.start); + t.send("garbage"); + t.send(frames.media); + await waitFor(() => raws.length === 3); + t.close(); + expect(raws[0]).toContain('"eventType":"start"'); + expect(raws[1]).toBe("garbage"); + expect(raws[2]).toContain('"eventType":"media"'); + }); +}); diff --git a/test/streams-wire.test.ts b/test/streams-wire.test.ts index 4f0b821..6b624f5 100644 --- a/test/streams-wire.test.ts +++ b/test/streams-wire.test.ts @@ -121,7 +121,7 @@ describe("start message", () => { // Shape per the StartStream docs' start-event example. const bwStart = { eventType: "start", - metadata: { accountId: "9900778", callId: "c-abc", to: "+15550001111", from: "+15550002222" }, + metadata: { accountId: "5555555", callId: "c-abc", to: "+15550001111", from: "+15550002222" }, streamParams: { callSid: "CA123", tenant: "acme" }, }; const port = nextPort(); @@ -192,6 +192,31 @@ describe("start message", () => { }); }); +// ─── bot connect timeout ──────────────────────────────────────────────────── + +describe("bot connect timeout", () => { + it("ready() rejects when the bot accepts TCP but never completes the handshake", async () => { + const { createServer } = await import("node:net"); + const blackHole = createServer((sock) => sock.on("error", () => {})); + await new Promise((r) => blackHole.listen(0, "127.0.0.1", r)); + const port = (blackHole.address() as import("node:net").AddressInfo).port; + + const source = new FakeBwSource(); + const bridge = new TwilioStreamBridge({ + botUrl: `ws://127.0.0.1:${port}`, + callSid: "CAhole", + accountSid: "AChole", + source, + connectTimeoutMs: 200, + }); + const startedAt = Date.now(); + await expect(bridge.ready()).rejects.toThrow(/handshake/i); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(150); + bridge.close(); + blackHole.close(); + }); +}); + // ─── stop message ─────────────────────────────────────────────────────────── describe("stop message", () => { diff --git a/test/translate-callback-auth.test.ts b/test/translate-callback-auth.test.ts index b161f32..b76ff14 100644 --- a/test/translate-callback-auth.test.ts +++ b/test/translate-callback-auth.test.ts @@ -17,6 +17,19 @@ describe("callback auth stamping", () => { const { bxml } = translateTwiml(`https://c.test/next`, { rewriteUrl, callbackAuth: auth }); expect(bxml).toContain(`username="bw-user"`); }); + it("adds destinationUsername/destinationPassword to a StartStream (Bandwidth sends them on the WebSocket upgrade)", () => { + const { bxml } = translateTwiml( + ``, + { rewriteUrl: (u) => `wss://translator.test/bw/stream?dest=${encodeURIComponent(u)}`, callbackAuth: auth }, + ); + expect(bxml).toMatch(/]*destinationUsername="bw-user"[^>]*destinationPassword="bw-pass"/); + // The StopStream that follows has no URL and gets nothing. + expect(bxml).toMatch(//); + }); + it("omits destination credentials from StartStream when callbackAuth is not provided", () => { + const { bxml } = translateTwiml(``, { rewriteUrl }); + expect(bxml).not.toContain("destinationUsername="); + }); it("omits credentials when callbackAuth is not provided", () => { const { bxml } = translateTwiml(`https://c.test/next`, { rewriteUrl }); expect(bxml).not.toContain("username=");