import type { VoiceAudioInput } from "@cloudflare/voice/client"; import { TelnyxRTC } from "@telnyx/webrtc"; import { float32ToInt16, computeRMS, PCM_CAPTURE_PROCESSOR_SOURCE, PCM_PLAYBACK_PROCESSOR_SOURCE } from "../audio/utils.js"; /** * Configuration for the TelnyxCallBridge. * * Uses JWT authentication (browser-side). The JWT is generated * server-side from a Telnyx API key + credential connection. */ export interface TelnyxCallBridgeConfig { /** JWT token from the Telnyx telephony credentials API. */ loginToken: string; /** Automatically answer inbound calls. @default false */ autoAnswer?: boolean; /** Enable debug logging. @default false */ debug?: boolean; } interface TelnyxCallLike { state?: string; answer?: () => void; hangup?: () => void; dtmf?: (digits: string) => void; remoteStream?: MediaStream | null; peer?: { instance?: RTCPeerConnection }; } interface TelnyxNotificationLike { type?: string; call?: TelnyxCallLike; } interface TelnyxRTCWithCalls { newCall: (options: { destinationNumber: string; callerNumber?: string; }) => TelnyxCallLike; } interface AudioInboundRtpStats extends RTCStats { kind?: string; bytesReceived?: number; packetsReceived?: number; packetsLost?: number; packetsDiscarded?: number; totalSamplesReceived?: number; jitterBufferEmittedCount?: number; } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } function isTelnyxNotification(value: unknown): value is TelnyxNotificationLike { return isRecord(value); } function hasNewCall( client: TelnyxRTC ): client is TelnyxRTC & TelnyxRTCWithCalls { return "newCall" in client && typeof client.newCall === "function"; } function getPeerConnection( call: TelnyxCallLike ): RTCPeerConnection | undefined { return call.peer?.instance; } function getRemoteStream(call: TelnyxCallLike): MediaStream | null { return call.remoteStream ?? null; } /** * Bridges Telnyx phone calls into the Cloudflare voice pipeline. * * Implements `VoiceAudioInput` from @cloudflare/voice — extracts PCM * audio from inbound phone calls and feeds it to the AI pipeline. * Also provides `playAudio()` for injecting response audio back * into the phone call. * * Usage: * ```typescript * const bridge = new TelnyxCallBridge({ loginToken: jwt }); * const voiceClient = new VoiceClient({ * agent: "my-agent", * audioInput: bridge, * }); * ``` */ export class TelnyxCallBridge implements VoiceAudioInput { // VoiceAudioInput callbacks — set by VoiceClient before start() onAudioLevel: ((rms: number) => void) | null = null; onAudioData?: ((pcm: ArrayBuffer) => void) | null = null; private readonly config: TelnyxCallBridgeConfig; private _connected = false; private _activeCall: TelnyxCallLike | null = null; private client: TelnyxRTC | null = null; private captureContext: AudioContext | null = null; private captureSource: MediaStreamAudioSourceNode | null = null; private captureWorklet: AudioWorkletNode | null = null; private captureBlobUrl: string | null = null; private captureAudioEl: HTMLAudioElement | null = null; private statsInterval: ReturnType | null = null; private playbackContext: AudioContext | null = null; private playbackWorklet: AudioWorkletNode | null = null; private playbackBlobUrl: string | null = null; private startPromise: Promise | null = null; private finishStart: (() => void) | null = null; private startAttempt = 0; private mediaSetupAttempt = 0; constructor(config: TelnyxCallBridgeConfig) { this.config = config; } /** Whether the Telnyx client is connected to the platform. */ get connected(): boolean { return this._connected; } /** The currently active Telnyx call, or null. */ get activeCall(): unknown | null { return this._activeCall; } /** Connect to Telnyx and start listening for calls. */ async start(): Promise { if (this._connected) return; if (this.startPromise) return this.startPromise; const attempt = ++this.startAttempt; const client = new TelnyxRTC({ login_token: this.config.loginToken, debug: this.config.debug }); this.client = client; this.startPromise = new Promise((resolve, reject) => { this.finishStart = resolve; client.on("telnyx.ready", () => { if (this.startAttempt !== attempt || this.client !== client) { resolve(); return; } this._connected = true; resolve(); }); client.on("telnyx.error", (error: unknown) => { if (this.startAttempt !== attempt || this.client !== client) { resolve(); return; } reject(error); }); client.on("telnyx.notification", (notification: unknown) => { if (this.startAttempt !== attempt || this.client !== client) return; this.handleNotification(notification); }); client.connect(); }).finally(() => { if (this.startAttempt === attempt) { this.startPromise = null; this.finishStart = null; } }); return this.startPromise; } /** Answer the current inbound call. */ answer(): void { if (!this._activeCall) throw new Error("No active call"); this._activeCall.answer?.(); } /** End the active call. */ hangup(): void { if (!this._activeCall) return; this._activeCall.hangup?.(); } /** * Initiate an outbound PSTN call. * @param destination Phone number or SIP URI to call. * @param callerNumber The caller ID number to present. * @returns The Telnyx Call object. */ dial(destination: string, callerNumber?: string): unknown { if (!this.client) throw new Error("Not connected — call start() first"); if (!hasNewCall(this.client)) { throw new Error("Telnyx client does not expose newCall()"); } const call = this.client.newCall({ destinationNumber: destination, callerNumber }); this._activeCall = call; return call; } /** Send DTMF digits on the active call. */ sendDTMF(digits: string): void { if (!this._activeCall) throw new Error("No active call"); this._activeCall.dtmf?.(digits); } /** * Clear any buffered audio in the playback pipeline. * Used during interrupt detection to stop stale audio from playing. */ clearPlaybackBuffer(): void { this.playbackWorklet?.port.postMessage("clear"); } /** * Inject PCM audio into the active phone call (agent → caller). * Accepts 16kHz mono Int16 PCM. Upsamples to 48kHz for WebRTC. * No-op if no active call. */ playAudio(pcm: ArrayBuffer): void { if (!this.playbackWorklet) return; const int16 = new Int16Array(pcm); // Upsample 16kHz → 48kHz (3x) via linear interpolation const upsampleRatio = 3; const float32 = new Float32Array(int16.length * upsampleRatio); for (let i = 0; i < int16.length; i++) { const current = int16[i] / 32768; const next = i < int16.length - 1 ? int16[i + 1] / 32768 : current; const base = i * upsampleRatio; for (let j = 0; j < upsampleRatio; j++) { float32[base + j] = current + (next - current) * (j / upsampleRatio); } } this.playbackWorklet.port.postMessage(float32); } /** Disconnect from Telnyx and clean up all resources. */ stop(): void { this.startAttempt++; this.mediaSetupAttempt++; this.finishStart?.(); this.finishStart = null; this.startPromise = null; this.stopAudioCapture(); this.stopAudioPlayback(); this._activeCall = null; this._connected = false; if (this.client) { this.client.disconnect(); this.client = null; } } private handleNotification(notification: unknown): void { if (!isTelnyxNotification(notification)) return; console.log( "[TelnyxCallBridge] notification:", notification.type, "call:", !!notification.call, "state:", notification.call?.state ); if (notification.type !== "callUpdate" || !notification.call) return; const call = notification.call; console.log("[TelnyxCallBridge] call state:", call.state); switch (call.state) { case "ringing": this._activeCall = call; if (this.config.autoAnswer) { console.log("[TelnyxCallBridge] auto-answering call"); call.answer?.(); } break; case "active": this._activeCall = call; console.log( "[TelnyxCallBridge] call active — starting audio capture + playback" ); this.mediaSetupAttempt++; this.stopAudioCapture(); this.stopAudioPlayback(); this.startAudioCapture(call, this.mediaSetupAttempt).catch((err) => console.error("[TelnyxCallBridge] startAudioCapture failed:", err) ); this.startAudioPlayback(call, this.mediaSetupAttempt).catch((err) => console.error("[TelnyxCallBridge] startAudioPlayback failed:", err) ); break; case "hangup": case "destroy": case "purge": this.mediaSetupAttempt++; this.stopAudioCapture(); this.stopAudioPlayback(); this._activeCall = null; break; } } private isCurrentMediaSetup(setupAttempt: number): boolean { return this.mediaSetupAttempt === setupAttempt; } private cleanupCaptureResources(resources: { audioEl?: HTMLAudioElement | null; context?: AudioContext | null; source?: MediaStreamAudioSourceNode | null; worklet?: AudioWorkletNode | null; blobUrl?: string | null; }): void { resources.worklet?.disconnect(); resources.source?.disconnect(); resources.context?.close(); if (resources.blobUrl) URL.revokeObjectURL(resources.blobUrl); if (resources.audioEl) { resources.audioEl.pause(); resources.audioEl.srcObject = null; resources.audioEl.remove(); } } private cleanupPlaybackResources(resources: { context?: AudioContext | null; worklet?: AudioWorkletNode | null; blobUrl?: string | null; }): void { resources.worklet?.disconnect(); resources.context?.close(); if (resources.blobUrl) URL.revokeObjectURL(resources.blobUrl); } private async startAudioCapture( call: TelnyxCallLike, setupAttempt: number ): Promise { // Get the remote audio track from the peer connection receiver. const pc = getPeerConnection(call); let track: MediaStreamTrack | null = null; if (pc) { const receivers = pc.getReceivers(); const audioReceiver = receivers.find( (r: RTCRtpReceiver) => r.track?.kind === "audio" ); track = audioReceiver?.track ?? null; } // Fall back to call.remoteStream if (!track) { const stream = getRemoteStream(call); track = stream?.getAudioTracks()?.[0] ?? null; } if (!track || track.readyState !== "live") { console.warn( "[TelnyxCallBridge] No live audio track — audio capture skipped" ); return; } // Ensure the track is enabled track.enabled = true; const remoteStream = new MediaStream([track]); // Attach the remote stream to an