app.js

import express from "express";
import { createServer } from "http";
import { WebSocketServer } from "ws";

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());

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 TELNYX_ALPHA_SENDER  = process.env.TELNYX_ALPHA_SENDER || TELNYX_FROM_NUMBER; 
const TELNYX_MESSAGING_PROFILE_ID = process.env.TELNYX_MESSAGING_PROFILE_ID;

const ACTIVE_VOICE = "Enceladus_3";
const log = (tag, ...args) => console.log(`[${new Date().toISOString()}] [${tag}]`, ...args);

let pendingCalls = new Map(); 
let activeLeadIds = new Map(); 
const incomingCallDebounce = new Map(); 

const MY_CELL_NUMBER = "+2347066923490"; // <-- Your agent phone number

// ── 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);
}

const sendTelnyxCmd = async (callId, action, body = {}) => {
  try {
    const url = callId ? `https://api.telnyx.com/v2/calls/${callId}/actions/${action}` : `https://api.telnyx.com/v2/calls`;
    const response = await fetch(url, {
      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()}`);
    return response;
  } catch (e) { log("TELNYX CMD NETWORK ERROR", e.message); }
};

app.post("/api/call-action", async (req, res) => {
  const { call_control_id, action, lead_id } = req.body;
  if (!pendingCalls.has(call_control_id)) return res.status(404).json({ error: "Call not pending" });
  if (lead_id) activeLeadIds.set(call_control_id, lead_id);
  pendingCalls.set(call_control_id, action);
  
  if (action === "reject") {
    await sendTelnyxCmd(call_control_id, "hangup");
    pendingCalls.delete(call_control_id);
  } else {
    await sendTelnyxCmd(call_control_id, "answer");
  }
  res.json({ ok: true });
});

app.post("/api/hangup", async (req, res) => {
  await sendTelnyxCmd(req.body.call_control_id, "hangup");
  res.json({ ok: true });
});

app.post("/api/send-sms", async (req, res) => {
  log("SMS", `Attempting outbound SMS to ${req.body.to}...`);
  try {
     const sendRequest = async (fromNumber) => {
         const payload = { from: fromNumber, to: req.body.to, text: req.body.text };
         if (TELNYX_MESSAGING_PROFILE_ID) payload.messaging_profile_id = TELNYX_MESSAGING_PROFILE_ID;
         return await fetch("https://api.telnyx.com/v2/messages", { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${TELNYX_API_KEY}` }, body: JSON.stringify(payload) });
     };

     let response = await sendRequest(TELNYX_ALPHA_SENDER);
     let data = await response.json();

     if (!response.ok && data.errors && data.errors.some(e => e.code === '40305' || e.code === '40012')) {
         log("SMS WARNING", `Alpha Sender rejected. Falling back to standard US number ${TELNYX_FROM_NUMBER}...`);
         response = await sendRequest(TELNYX_FROM_NUMBER);
         data = await response.json();
     }

     if (!response.ok) log("SMS ERROR", JSON.stringify(data));
     else log("SMS SUCCESS", `Message accepted by Telnyx for delivery to ${req.body.to}`);
     res.json(data);
  } catch(e) { res.status(500).json({ error: e.message }); }
});

app.post("/dial", async (req, res) => {
  const { to, lead_id, action = "manual" } = req.body;
  
  try {
    let callPayload = { connection_id: TELNYX_CONNECTION_ID, from: TELNYX_FROM_NUMBER };

    if (action === "manual") {
      log("DIAL", `[Manual] Dialing Agent First (${MY_CELL_NUMBER})...`);
      // We dial YOU first. We pass the customer's number securely in the client_state.
      const stateObj = { type: "agent_first", customer_number: to, lead_id: lead_id };
      callPayload.to = MY_CELL_NUMBER;
      callPayload.client_state = Buffer.from(JSON.stringify(stateObj)).toString("base64");
    } else {
      log("DIAL", `[AI] Dialing Customer directly (${to})...`);
      // Normal AI dialing directly to customer
      callPayload.to = to;
    }

    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(callPayload)
    });
    const data = await response.json();
    if (!response.ok) return res.status(400).json({ ok: false, error: data?.errors?.[0]?.detail });

    const ccid = data.data.call_control_id;
    pendingCalls.set(ccid, action); 
    if (lead_id) activeLeadIds.set(ccid, lead_id);

    await fetch(`${AI_SERVER_URL}/api/log-outbound`, {
      method: "POST", headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ call_control_id: ccid, action, lead_id })
    }).catch(() => {});

    res.json({ ok: true, call_control_id: ccid });
  } catch (err) { res.status(500).json({ ok: false, error: err.message }); }
});

app.post("/webhook/call", async (req, res) => {
  res.sendStatus(200); 
  const type    = req.body?.data?.event_type;
  const payload = req.body?.data?.payload;
  const callId  = payload?.call_control_id;
  if (!type) return;

  switch (type) {
    case "call.initiated":
      if (payload?.direction === "incoming") {
        const fromNumber = payload.from;
        const now = Date.now();
        if (incomingCallDebounce.has(fromNumber) && now - incomingCallDebounce.get(fromNumber) < 10000) return;
        incomingCallDebounce.set(fromNumber, now);

        log("INCOMING", `From: ${fromNumber}`);
        pendingCalls.set(callId, "waiting");
        await fetch(`${AI_SERVER_URL}/api/incoming-call`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ call_control_id: callId, from: fromNumber }) }).catch(()=>{});
      }
      break;

    case "call.answered":
      // 1. CHECK IF THIS IS A BRIDGING STATE (Agent picked up, or Customer picked up)
      if (payload?.client_state) {
        try {
          const stateStr = Buffer.from(payload.client_state, "base64").toString("utf8");
          const stateObj = JSON.parse(stateStr);

          // SCENARIO A: Agent just answered their cell phone.
          if (stateObj.type === "agent_first") {
            log("BRIDGING", `Agent picked up! Dialing Customer (${stateObj.customer_number})...`);
            
            // Start recording the agent leg so we capture the whole conversation
            await sendTelnyxCmd(callId, "record_start", { format: "mp3", channels: "dual", play_beep: false });
            
            // Play ringing to the AGENT so they know the customer is being called
            await sendTelnyxCmd(callId, "playback_start", { audio_url: "https://actions.telnyx.com/sample-assets/ringback.wav", target_legs: "self" });

            // Now dial the CUSTOMER, passing the Agent's call ID forward
            const nextState = { type: "customer_answered", leg_a_id: callId };
            await sendTelnyxCmd(null, "dial", {
              to: stateObj.customer_number, 
              from: TELNYX_FROM_NUMBER,
              connection_id: TELNYX_CONNECTION_ID, 
              client_state: Buffer.from(JSON.stringify(nextState)).toString("base64")
            });
            return; 
          }

          // SCENARIO B: Customer just answered the phone.
          if (stateObj.type === "customer_answered") {
            log("BRIDGING", `Customer picked up! Bridging them to the Agent...`);
            
            // Stop the ringing in the Agent's ear
            await sendTelnyxCmd(stateObj.leg_a_id, "playback_stop");
            
            // Bridge the Customer (callId) to the Agent (leg_a_id)
            await sendTelnyxCmd(callId, "bridge", { call_control_id: stateObj.leg_a_id });
            return;
          }
        } catch (e) { log("STATE ERROR", "Failed to parse client state JSON."); }
      }

      // 2. IF NOT BRIDGING, IT IS AN AI CALL OR INCOMING CALL
      const actionType = pendingCalls.get(callId) || "ai"; 
      if (actionType === "ai") {
        log("AI CONNECTING", `Starting stream for ${callId}`);
        await sendTelnyxCmd(callId, "record_start", { format: "mp3", channels: "dual", play_beep: false });
        await sendTelnyxCmd(callId, "playback_start", { audio_url: `http://${req.headers.host}/ambience.mp3`, target_legs: "caller", loop: "infinity" });
        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" });
        setTimeout(async () => { log("HANGUP", `Enforcing 90s limit on ${callId}`); await sendTelnyxCmd(callId, "hangup"); }, 90000);
      }
      break;

    case "call.transcription":
      if (!payload?.transcription_data?.is_final) break;
      try {
        const aiRes = await fetch(`${AI_SERVER_URL}/api/process-text-turn`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ text: payload.transcription_data.transcript.trim(), call_id: callId, lead_id: activeLeadIds.get(callId) || null }) });
        if (aiRes.ok) {
          const aiData = await aiRes.json();
          if (aiData.speak_url) 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":
      if (payload.recording_urls?.mp3) {
        await fetch(`${AI_SERVER_URL}/api/save-recording`, {
          method: "POST", headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ call_control_id: callId, recording_url: payload.recording_urls.mp3 })
        }).then(async (r) => { if (!r.ok) log("RECORDING ERROR", `AI Server status: ${r.status}`); }).catch((err) => log("RECORDING NET ERROR", err.message));
      }
      break;

    case "call.hangup":
      log("HANGUP", `Call ended. Reason: ${payload.hangup_cause}`); 
      pendingCalls.delete(callId);
      activeLeadIds.delete(callId);
      await fetch(`${AI_SERVER_URL}/api/call-ended`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ call_control_id: callId }) }).catch(() => {});
      break;
  }
});

wss.on("connection", (ws) => {
  let isSpeaking = false;
  let silenceFrames = 0;
  let callControlId = null;
  let hasPlayedFiller = false;
  let hasGreeted = false;
  let totalNoiseFrames = 0;

  ws.on("message", async (raw) => {
    try {
      const msg = JSON.parse(raw);
      if (msg.event === "start") {
        callControlId = msg.start?.call_control_id;
        setTimeout(async () => {
          if (hasGreeted || !callControlId) return;
          hasGreeted = true;
          const greetingPhrase = "Hi, this is Nathan.";
          await sendTelnyxCmd(callControlId, "playback_start", { audio_url: `${AI_SERVER_URL}/play-audio?voice=${ACTIVE_VOICE}&phrase=${encodeURIComponent(greetingPhrase)}`, target_legs: "self" });
          fetch(`${AI_SERVER_URL}/api/register-greeting`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ call_id: callControlId, phrase: greetingPhrase, lead_id: activeLeadIds.get(callControlId) || null }) }).catch(() => {});
        }, 1500);
      }

      if (msg.event === "media") {
        const pcm16 = decodeMuLaw(Buffer.from(msg.media.payload, "base64"));
        const rms = calculateRMS(pcm16);

        if (rms > 500) {
          hasGreeted = true;
          silenceFrames = 0;
          totalNoiseFrames++;
          if (!isSpeaking) {
            isSpeaking = true;
            hasPlayedFiller = false;
            await sendTelnyxCmd(callControlId, "playback_stop"); 
          }
        } else if (isSpeaking) {
          silenceFrames++;
          if (silenceFrames > 35 && !hasPlayedFiller) {
            hasPlayedFiller = true;
            isSpeaking = false;
            if (totalNoiseFrames > 40) {
              const fillers = ["Right.", "Got it.", "Okay.", "Yeah.", "Ah, okay."];
              sendTelnyxCmd(callControlId, "playback_start", { audio_url: `${AI_SERVER_URL}/play-audio?voice=${ACTIVE_VOICE}&phrase=${encodeURIComponent(fillers[Math.floor(Math.random() * fillers.length)])}`, target_legs: "self" }).catch(() => {});
            }
            totalNoiseFrames = 0;
          }
        }
      }
    } catch {}
  });
});

server.listen(PORT, "0.0.0.0", () => {
  log("STARTUP", `Telecom routing server running on port ${PORT}`);
});