import express from "express";
import { createServer } from "http";
import { WebSocketServer } from "ws";
import { randomUUID } from "crypto";
const app = express();
const server = createServer(app);
const wss = new WebSocketServer({ server, path: "/audio-stream" });
app.use(express.static("public", { index: false }));
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
const PORT = process.env.PORT || 7860;
const TELNYX_API_KEY = process.env.TELNYX_API_KEY;
const TELNYX_CONNECTION_ID = process.env.TELNYX_CONNECTION_ID;
const TELNYX_FROM_NUMBER = process.env.TELNYX_FROM_NUMBER;
const AI_SERVER_URL = process.env.AI_SERVER_URL;
const ACTIVE_VOICE = "Enceladus_3";
const log = (tag, ...args) => console.log(`[${new Date().toISOString()}] [${tag}]`, ...args);
let callLogs = [];
const sendTelnyxCmd = async (callId, action, body = {}) => {
if (action === "playback_start") {
updateLog(callId, { is_playing: true, playback_started_at: Date.now() });
}
try {
const response = await fetch(`https://api.telnyx.com/v2/calls/${callId}/actions/${action}`, {
method: "POST",
headers: { "Content-Type": "application/json", Authorization: `Bearer ${TELNYX_API_KEY}` },
body: JSON.stringify(body)
});
if (!response.ok) log("TELNYX CMD ERROR", `[${action}] ${await response.text()}`);
else log("TELNYX CMD SUCCESS", `[${action}] accepted.`);
return response;
} catch (e) { log("TELNYX CMD NETWORK ERROR", e.message); }
};
const updateLog = (callId, updates) => {
const entry = callLogs.find(l => l.call_control_id === callId);
if (entry) Object.assign(entry, updates);
return entry;
};
// ── MU-LAW VAD ──
const MU_LAW_TABLE = new Int16Array(256);
for (let i = 0; i < 256; i++) {
const sign = (i & 0x80) ? -1 : 1;
const exponent = (~i >> 4) & 0x07;
const mantissa = ~i & 0x0f;
const sample = ((mantissa << 3) + 132) << exponent;
MU_LAW_TABLE[i] = sign * (sample - 132);
}
function decodeMuLaw(buffer) {
const out = new Int16Array(buffer.length);
for (let i = 0; i < buffer.length; i++) out[i] = MU_LAW_TABLE[buffer[i]];
return out;
}
function calculateRMS(pcm) {
let sum = 0;
for (let i = 0; i < pcm.length; i++) sum += pcm[i] * pcm[i];
return Math.sqrt(sum / pcm.length);
}
// ── DASHBOARD ──
app.get("/", (req, res) => {
res.send(`
Elite Telecom Control Panel
`);
});
app.get("/api/logs", (req, res) => {
res.json(callLogs.map(({ timers, ...e }) => e).reverse());
});
app.get("/health", (req, res) => res.json({ status: "ok" }));
app.get("/play-recording/:ccid", async (req, res) => {
const entry = callLogs.find(l => l.call_control_id === req.params.ccid);
if (!entry?.raw_telnyx_url) return res.sendStatus(404);
try {
const s3 = await fetch(entry.raw_telnyx_url);
if (!s3.ok) return res.sendStatus(s3.status);
const buf = Buffer.from(await s3.arrayBuffer());
res.setHeader("Content-Type", "audio/mpeg");
res.setHeader("Content-Length", buf.length);
res.send(buf);
} catch { res.sendStatus(500); }
});
app.post("/dial", async (req, res) => {
const { to } = req.body;
log("DIAL", `Dialing ${to}...`);
try {
const response = await fetch("https://api.telnyx.com/v2/calls", {
method: "POST",
headers: { "Content-Type": "application/json", Authorization: `Bearer ${TELNYX_API_KEY}` },
body: JSON.stringify({
connection_id: TELNYX_CONNECTION_ID,
to, from: TELNYX_FROM_NUMBER,
custom_headers: [{ name: "X-Voice-Selection", value: ACTIVE_VOICE }]
})
});
const data = await response.json();
if (!response.ok) return res.status(400).json({ ok: false, error: data?.errors?.[0]?.detail });
callLogs.push({
id: randomUUID(), call_control_id: data.data.call_control_id,
target_number: to, voice_used: ACTIVE_VOICE, status: "dialing",
last_transcript: "", recording_url: null, raw_telnyx_url: null,
is_playing: false, playback_started_at: null,
answered_at: null, timers: {}, created_at: new Date().toISOString()
});
res.json({ ok: true });
} catch (err) { res.status(500).json({ ok: false, error: err.message }); }
});
app.post("/call/hangup", async (req, res) => {
const { call_control_id } = req.body;
if (!call_control_id) return res.status(400).json({ error: "Missing call_control_id" });
await sendTelnyxCmd(call_control_id, "hangup");
res.json({ ok: true });
});
// ── WEBHOOK ──
app.post("/webhook/call", async (req, res) => {
const type = req.body?.data?.event_type;
const payload = req.body?.data?.payload;
const callId = payload?.call_control_id;
if (!type) return res.sendStatus(200);
const entry = callLogs.find(l => l.call_control_id === callId);
if (entry?.status === "completed" && type !== "call.recording.saved") return res.sendStatus(200);
switch (type) {
case "call.initiated":
log("INITIATED", `dir=${payload?.direction}`);
if (payload?.direction === "incoming") {
callLogs.push({
id: randomUUID(), call_control_id: callId, target_number: payload?.from,
voice_used: ACTIVE_VOICE, status: "in-progress", last_transcript: "",
recording_url: null, raw_telnyx_url: null, is_playing: false,
playback_started_at: null, answered_at: null, timers: {}, created_at: new Date().toISOString()
});
await sendTelnyxCmd(callId, "answer");
}
break;
case "call.answered":
log("ANSWERED", `Call ${callId} connected.`);
updateLog(callId, { status: "in-progress", answered_at: new Date().toISOString() });
await sendTelnyxCmd(callId, "record_start", { format: "mp3", channels: "dual", play_beep: false });
await sendTelnyxCmd(callId, "streaming_start", {
stream_url: `wss://${req.headers.host}/audio-stream`,
stream_track: "inbound_track"
});
await sendTelnyxCmd(callId, "transcription_start", {
transcription_engine: "Telnyx",
transcription_tracks: "inbound"
});
if (entry) {
entry.timers.hangup = setTimeout(async () => {
if (entry.status === "completed") return;
log("HANGUP", "Enforcing 90-second limit.");
await sendTelnyxCmd(callId, "hangup");
}, 90000);
}
break;
case "call.playback.ended":
case "call.playback.stopped":
updateLog(callId, { is_playing: false });
break;
case "call.transcription":
const tData = payload?.transcription_data;
if (!tData) break;
const voice = entry?.voice_used || ACTIVE_VOICE;
if (!tData.is_final && entry?.is_playing) {
log("BARGE-IN", "User speaking — stopping AI.");
updateLog(callId, { is_playing: false });
await sendTelnyxCmd(callId, "playback_stop");
}
if (tData.is_final) {
const transcript = tData.transcript.trim();
if (!transcript) break;
log("TRANSCRIPT", `"${transcript}"`);
updateLog(callId, { last_transcript: transcript });
try {
const aiRes = await fetch(`${AI_SERVER_URL}/api/process-text-turn`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ text: transcript, voice, call_id: callId })
});
if (aiRes.ok) {
const aiData = await aiRes.json();
if (aiData.speak_url && entry?.status !== "completed") {
log("AI_REPLY", `"${aiData.raw_text}"`);
await sendTelnyxCmd(callId, "playback_start", {
audio_url: aiData.speak_url, target_legs: "self"
});
}
}
} catch (e) { log("AI_SERVER_ERROR", e.message); }
}
break;
case "call.recording.saved":
const rawUrl = payload.recording_urls.mp3;
log("RECORDING", rawUrl);
updateLog(callId, { raw_telnyx_url: rawUrl, recording_url: `/play-recording/${callId}` });
break;
case "call.hangup":
updateLog(callId, { status: "completed", is_playing: false });
if (entry?.timers) Object.values(entry.timers).forEach(clearTimeout);
log("CALL ENDED", "Terminated.");
break;
}
res.sendStatus(200);
});
// ── WEBSOCKET: VAD + ECHO SUPPRESSION ──
wss.on("connection", (ws) => {
log("WS", "VAD WebSocket connected");
let isSpeaking = false;
let silenceFrames = 0;
let callControlId = null;
let hasPlayedFiller = false;
let hasGreeted = false;
let totalNoiseFrames = 0;
let wasAiPlaying = false; // Tracks AI playing state across frames for transition detection
const SILENCE_THRESHOLD_FRAMES = 35; // 700ms
const VOLUME_THRESHOLD = 500;
const ECHO_GRACE_PERIOD = 2000; // 2s after AI starts playing, ignore inbound volume
ws.on("message", async (raw) => {
try {
const msg = JSON.parse(raw);
if (msg.event === "start") {
callControlId = msg.start?.call_control_id;
// Greeting fires 1.5s after call connects.
// Uses "Hi, this is Nathan." then registers the turn to the AI server
// so the LLM conversation history knows the intro already happened.
setTimeout(async () => {
if (hasGreeted || !callControlId) return;
const e = callLogs.find(l => l.call_control_id === callControlId);
if (e?.status !== "in-progress") return;
hasGreeted = true;
const greetingPhrase = "Hi, this is Nathan.";
log("SMART_GREETING", `Playing: "${greetingPhrase}"`);
await sendTelnyxCmd(callControlId, "playback_start", {
audio_url: `${AI_SERVER_URL}/play-audio?voice=${e.voice_used}&phrase=${encodeURIComponent(greetingPhrase)}`,
target_legs: "self"
});
// Register greeting into conversation history on the AI server
fetch(`${AI_SERVER_URL}/api/register-greeting`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ call_id: callControlId, phrase: greetingPhrase })
}).catch(() => {});
}, 1500);
}
if (msg.event === "media") {
const pcm16 = decodeMuLaw(Buffer.from(msg.media.payload, "base64"));
const rms = calculateRMS(pcm16);
const entry = callLogs.find(l => l.call_control_id === callControlId);
const nowAiPlaying = entry?.is_playing || false;
const timeSincePlayback = entry?.playback_started_at
? (Date.now() - entry.playback_started_at) : 9999;
const isEchoWindow = timeSincePlayback < ECHO_GRACE_PERIOD;
// ── When AI starts speaking, hard-reset VAD state ──
// This prevents accumulated silence/noise frames from causing a phantom filler
if (!wasAiPlaying && nowAiPlaying) {
isSpeaking = false;
silenceFrames = 0;
totalNoiseFrames = 0;
hasPlayedFiller = false;
}
wasAiPlaying = nowAiPlaying;
if (rms > VOLUME_THRESHOLD) {
// Suppress echo: if AI just started talking, ignore inbound spikes
if (isEchoWindow) return;
hasGreeted = true;
silenceFrames = 0;
totalNoiseFrames++;
if (!isSpeaking) {
isSpeaking = true;
hasPlayedFiller = false;
if (entry?.is_playing) {
log("BARGE-IN (VAD)", "User speaking — stopping AI.");
entry.is_playing = false;
await sendTelnyxCmd(callControlId, "playback_stop");
}
}
} else if (isSpeaking) {
silenceFrames++;
if (silenceFrames > SILENCE_THRESHOLD_FRAMES && !hasPlayedFiller) {
hasPlayedFiller = true;
isSpeaking = false;
// 8-second blackout: don't fire fillers during IVR intro
const callAge = entry?.answered_at
? Date.now() - new Date(entry.answered_at).getTime() : 0;
if (totalNoiseFrames > 40 && callAge > 8000) {
log("VAD_EARLY_TURN", "Long pause detected — playing filler.");
const fillers = ["Right.", "Got it.", "Okay.", "Yeah.", "Ah, okay."];
const filler = fillers[Math.floor(Math.random() * fillers.length)];
sendTelnyxCmd(callControlId, "playback_start", {
audio_url: `${AI_SERVER_URL}/play-audio?voice=${entry?.voice_used || ACTIVE_VOICE}&phrase=${encodeURIComponent(filler)}`,
target_legs: "self"
}).catch(() => {});
}
totalNoiseFrames = 0;
}
}
}
} catch {}
});
});
server.listen(PORT, "0.0.0.0", () => {
log("STARTUP", `Telecom Server running on port ${PORT}`);
});