// VMO Media Bridge v5 — direct-bridge AI + live-audio Listen fanout
// ----------------------------------------------------------------------------
// /media   : Twilio Media Streams ↔ OpenAI Realtime (direct-bridge AI receptionist)
// /listen  : Browser listeners receive μ-law frames (caller + AI) for live monitor
// /healthz : liveness probe
//
// Media token contract: HMAC-SHA256(VMO_MEDIA_SECRET || VMO_LISTEN_SECRET, "<call_sid>|<user_id>|<voice>|<exp>")
// Twilio start.customParameters: call_sid, user_id, voice, media_exp, media_sig
// Listen token contract: HMAC-SHA256(VMO_LISTEN_SECRET, "<call_sid>|<user_id>|<exp>")
// Query params: ?call_sid=CA...&user_id=N&exp=<unix-epoch>&sig=<hex>
// ----------------------------------------------------------------------------
import 'dotenv/config';
import Fastify from 'fastify';
import websocketPlugin from '@fastify/websocket';
import WebSocket from 'ws';
import crypto from 'crypto';

const PORT = parseInt(process.env.PORT || '8080', 10);
const OPENAI_API_KEY = process.env.OPENAI_API_KEY;
const VMO_LISTEN_SECRET = process.env.VMO_LISTEN_SECRET || '';
const VMO_MEDIA_SECRET = process.env.VMO_MEDIA_SECRET || VMO_LISTEN_SECRET || '';
const OPENAI_MODEL = process.env.OPENAI_MODEL || 'gpt-realtime';

if (!OPENAI_API_KEY) {
  console.error('[FATAL] OPENAI_API_KEY missing in .env');
  process.exit(1);
}
if (!VMO_LISTEN_SECRET) {
  console.warn('[WARN] VMO_LISTEN_SECRET missing — /listen endpoint will reject all connections');
}
if (!VMO_MEDIA_SECRET) {
  console.warn('[WARN] VMO_MEDIA_SECRET missing — /media endpoint will reject all connections');
} else if (!process.env.VMO_MEDIA_SECRET) {
  console.warn('[WARN] VMO_MEDIA_SECRET missing — using VMO_LISTEN_SECRET for /media signing');
}

const SUPPORTED_VOICES = new Set([
  'alloy','ash','ballad','coral','echo','fable','marin','cedar','nova','onyx','sage','shimmer','verse'
]);
const DEFAULT_VOICE = 'ash';
const SYSTEM_INSTRUCTIONS = 'You are VMO, a friendly and professional AI receptionist. Greet callers warmly, listen, and help them. Keep replies short and natural.';
const MAX_PENDING_INPUT_FRAMES = 250;

// callSid -> { twilioWs, openaiWs, streamSid, voice, userId, listeners:Set, openAiReady, pendingDeltas:[] }
const sessions = new Map();

const fastify = Fastify({ logger: false });
await fastify.register(websocketPlugin, { options: { maxPayload: 1048576 } });

// ----------------------------------------------------------------------------
// Health
// ----------------------------------------------------------------------------
fastify.get('/healthz', async () => ({ ok: true, sessions: sessions.size, ts: Date.now() }));

// ----------------------------------------------------------------------------
// /media  — Twilio Media Stream ↔ OpenAI Realtime
// ----------------------------------------------------------------------------
fastify.get('/media', { websocket: true }, (conn, req) => {
  console.log('[Twilio] Incoming Media Stream Connected (awaiting signed start customParameters)');

  const twilioWs = conn.socket || conn;
  const session = {
    twilioWs,
    openaiWs: null,
    streamSid: null,
    callSid: null,
    voice: DEFAULT_VOICE,
    userId: 0,
    listeners: new Set(),
    mediaAuthorized: false,
    openAiSocketOpen: false,
    openAiReady: false,
    sessionUpdateSent: false,
    pendingInputAudio: [],
    pendingDeltas: [],
  };

  const rejectMedia = (code, msg) => {
    console.warn(`[Media] reject (${code}): ${msg} streamSid=${session.streamSid || '-'} callSid=${session.callSid || '-'}`);
    try { twilioWs.send(JSON.stringify({ event: 'error', code, message: msg })); } catch {}
    try { twilioWs.close(1008, msg); } catch {}
  };

  // -- Twilio side -----------------------------------------------------------
  twilioWs.on('message', (raw) => {
    let evt;
    try { evt = JSON.parse(raw.toString()); } catch { return; }

    switch (evt.event) {
      case 'connected':
        console.log('[Twilio] connected');
        break;
      case 'start': {
        session.streamSid = evt.start?.streamSid || null;
        const auth = validateMediaStart(evt.start || {});
        if (!auth.ok) {
          rejectMedia(auth.code, auth.message);
          return;
        }

        session.mediaAuthorized = true;
        session.callSid = auth.callSid;
        session.voice = auth.voice;
        session.userId = auth.userId;
        session.initialGreeting = auth.initialGreeting || '';
        session.initialInstruction = auth.initialInstruction || '';
        session.extensionMenu = auth.extensionMenu || '';
        session.extensionActionUrl = auth.extensionActionUrl || '';
        session.extensionCount = auth.extensionCount || 0;
        session.languagePolicy = auth.languagePolicy || '';
        session.defaultLanguage = auth.defaultLanguage || 'en';
        session.defaultLanguageName = auth.defaultLanguageName || 'English';
        sessions.set(session.callSid, session);
        console.log(`[Twilio] start authorized streamSid=${session.streamSid} callSid=${session.callSid} voice=${session.voice} user=${session.userId}`);

        connectOpenAI(session);
        break;
      }
      case 'media':
        if (evt.media?.payload) {
          if (!session.mediaAuthorized) {
            rejectMedia('not_authorized', 'Media received before signed start');
            return;
          }
          sendCallerAudio(session, evt.media.payload);
          // Fan out caller audio to live listeners
          fanout(session, 'caller', evt.media.payload);
        }
        break;
      case 'mark':
        // ignore
        break;
      case 'stop':
        console.log(`[Twilio] stop callSid=${session.callSid}`);
        try { session.openaiWs?.close(); } catch {}
        cleanup(session);
        break;
    }
  });

  twilioWs.on('close', () => {
    console.log(`[Twilio] WebSocket closed callSid=${session.callSid}`);
    try { session.openaiWs?.close(); } catch {}
    cleanup(session);
  });
  twilioWs.on('error', (err) => {
    console.error('[Twilio] WS error', err.message);
  });
});

function connectOpenAI(session) {
  if (session.openaiWs) return;

  const openaiWs = new WebSocket(
    `wss://api.openai.com/v1/realtime?model=${encodeURIComponent(OPENAI_MODEL)}`,
    { headers: { Authorization: `Bearer ${OPENAI_API_KEY}` } }
  );
  session.openaiWs = openaiWs;

  openaiWs.on('open', () => {
    console.log('[OpenAI] Connected after media auth');
    session.openAiSocketOpen = true;
    maybeSendSessionUpdate(session);
  });

  openaiWs.on('message', (raw) => {
    let evt;
    try { evt = JSON.parse(raw.toString()); } catch { return; }

    switch (evt.type) {
      case 'session.created':
        console.log('[OpenAI] session.created');
        break;
      case 'session.updated':
        console.log('[OpenAI] session.updated');
        session.openAiReady = true;
        flushPendingInputAudio(session);
        maybeSendGreeting(session);
        break;
      case 'response.output_audio.delta':
        if (evt.delta) {
          if (session.streamSid && session.twilioWs.readyState === 1) {
            session.twilioWs.send(JSON.stringify({
              event: 'media',
              streamSid: session.streamSid,
              media: { payload: evt.delta },
            }));
          } else {
            session.pendingDeltas.push(evt.delta);
          }
          fanout(session, 'ai', evt.delta);
        }
        break;
      case 'response.function_call_arguments.done': {
          try {
            const name = evt.name || '';
            const callId = evt.call_id || '';
            let args = {};
            try { args = JSON.parse(evt.arguments || '{}'); } catch (e) { args = {}; }
            if (name === 'ring_department' && session && session.extensionActionUrl) {
              const digit = String(args.digit || '').replace(/\D+/g, '');
              if (digit) {
                const url = session.extensionActionUrl
                  + (session.extensionActionUrl.includes('?') ? '&' : '?')
                  + 'digit=' + encodeURIComponent(digit);
                console.log('[Bridge] ring_department tool fired digit=' + digit + ' callSid=' + (session.callSid || ''));
                fetch(url, { method: 'POST' })
                  .then(r => r.text().then(t => console.log('[Bridge] extension_action_url status=' + r.status + ' body=' + t.slice(0, 200))))
                  .catch(err => console.error('[Bridge] extension_action_url ERROR: ' + err.message));
              } else {
                console.warn('[Bridge] ring_department called with empty digit');
              }
              // Tell OpenAI the function returned so it continues the conversation.
              try {
                openaiWs.send(JSON.stringify({
                  type: 'conversation.item.create',
                  item: {
                    type: 'function_call_output',
                    call_id: callId,
                    output: JSON.stringify({ success: true, digit: digit || null })
                  }
                }));
                openaiWs.send(JSON.stringify({ type: 'response.create' }));
              } catch (e) {
                console.error('[Bridge] function_call_output send failed: ' + e.message);
              }
            }
          } catch (e) {
            console.error('[Bridge] function_call handler error: ' + e.message);
          }
          break;
        }
        case 'response.output_audio.done':
        break;
      case 'response.done':
        if (session.streamSid && session.twilioWs.readyState === 1) {
          session.twilioWs.send(JSON.stringify({
            event: 'mark',
            streamSid: session.streamSid,
            mark: { name: 'response_done' },
          }));
        }
        break;
      case 'error':
        console.error('[OpenAI] error', JSON.stringify(evt));
        break;
    }
  });

  openaiWs.on('close', (code, reason) => {
    console.log(`[OpenAI] Closed ${code} ${reason?.toString() || ''}`);
    session.openAiSocketOpen = false;
    session.openAiReady = false;
    if (session.openaiWs === openaiWs) session.openaiWs = null;
  });
  openaiWs.on('error', (err) => {
    console.error('[OpenAI] socket error', err.message);
  });
}

function validateMediaStart(start) {
  if (!VMO_MEDIA_SECRET) {
    return { ok: false, code: 'no_secret', message: 'Bridge has no media secret configured' };
  }

  const customParameters = start.customParameters || {};
  const twilioCallSid = String(start.callSid || '').trim();
  const signedCallSid = String(customParameters.call_sid || '').trim();
  const callSid = signedCallSid || twilioCallSid;
  const userId = parseInt(customParameters.user_id || '0', 10);
  const requestedVoice = String(customParameters.voice || '').trim().toLowerCase();
  const voice = requestedVoice || DEFAULT_VOICE;
  const exp = parseInt(customParameters.media_exp || customParameters.exp || '0', 10);
  const sig = String(customParameters.media_sig || customParameters.sig || '').trim().toLowerCase();

  if (!callSid || !userId || !exp || !sig) {
    return { ok: false, code: 'bad_request', message: 'Missing signed media parameters' };
  }
  if (signedCallSid && twilioCallSid && signedCallSid !== twilioCallSid) {
    return { ok: false, code: 'call_sid_mismatch', message: 'Signed call SID does not match Twilio call SID' };
  }
  if (userId <= 0) {
    return { ok: false, code: 'bad_user', message: 'Invalid user ID' };
  }
  if (requestedVoice && !SUPPORTED_VOICES.has(requestedVoice)) {
    return { ok: false, code: 'bad_voice', message: 'Unsupported voice' };
  }

  const now = Math.floor(Date.now() / 1000);
  if (exp < now) {
    return { ok: false, code: 'expired', message: 'Media token expired' };
  }
  if (exp > now + 600) {
    return { ok: false, code: 'bad_exp', message: 'Media token expiry too far in future' };
  }

  const expected = crypto
    .createHmac('sha256', VMO_MEDIA_SECRET)
    .update(`${callSid}|${userId}|${voice}|${exp}`)
    .digest('hex');

  if (!timingSafeHexEqual(sig, expected)) {
    return { ok: false, code: 'bad_sig', message: 'Bad media signature' };
  }

  return {
    ok: true, callSid, userId, voice, exp,
    initialGreeting: String(customParameters.initial_greeting || '').trim(),
    initialInstruction: String(customParameters.initial_instruction || '').trim(),
    extensionMenu: String(customParameters.extension_menu || '').trim(),
    languagePolicy: String(customParameters.language_policy || '').trim(),
    defaultLanguage: String(customParameters.default_language || 'en').trim(),
    defaultLanguageName: String(customParameters.default_language_name || 'English').trim(),
  
    extensionActionUrl: String(customParameters.extension_action_url || customParameters.extensionActionUrl || '').trim(),
    extensionCount: parseInt(customParameters.extension_count || customParameters.extensionCount || '0', 10) || 0,
  };
}

function timingSafeHexEqual(actual, expected) {
  if (!/^[a-f0-9]{64}$/i.test(actual)) return false;
  const actualBuffer = Buffer.from(actual, 'hex');
  const expectedBuffer = Buffer.from(expected, 'hex');
  return actualBuffer.length === expectedBuffer.length && crypto.timingSafeEqual(actualBuffer, expectedBuffer);
}


function buildTools(session) {
  if (!session || !session.extensionActionUrl) return [];
  return [
    {
      type: 'function',
      name: 'ring_department',
      description: 'Ring the configured extension phone for the chosen department. Call this immediately when the caller selects a department by speaking its name or by saying/pressing its digit. The plugin will dial the matching phone number in parallel.',
      parameters: {
        type: 'object',
        properties: {
          digit: { type: 'string', description: 'The extension digit the caller chose, e.g. "1".' }
        },
        required: ['digit']
      }
    }
  ];
}

function buildInstructions(session) {
  const parts = [SYSTEM_INSTRUCTIONS];
  if (session.languagePolicy) parts.push(session.languagePolicy);
  if (session.extensionMenu) {
    parts.push(session.extensionMenu);
    parts.push('CRITICAL: When the caller chooses a department either by saying its name or by pressing/saying its digit, you MUST first invoke the ring_department tool with the matching digit BEFORE telling the caller that the department has been notified. Never claim a department has been notified unless you have just called the ring_department tool. After calling the tool, acknowledge their choice and keep helping them until that department joins.');
  }
  if (session.initialInstruction) parts.push(session.initialInstruction);
  return parts.join(' ');
}

function maybeSendSessionUpdate(session) {
  // Need both: openai socket open AND Twilio start received (so we know the voice)
  if (!session.openAiSocketOpen || !session.streamSid) return;
  if (session.sessionUpdateSent) return;
  session.sessionUpdateSent = true;
  console.log(`[Bridge] Sending OpenAI session.update voice=${session.voice}`);
  try {
    session.openaiWs.send(JSON.stringify({
      type: 'session.update',
      session: {
        type: 'realtime',
        model: OPENAI_MODEL,
        output_modalities: ['audio'],
        audio: {
          input: {
            format: { type: 'audio/pcmu' },
            turn_detection: { type: 'server_vad' },
          },
          output: {
            format: { type: 'audio/pcmu' },
            voice: session.voice,
          },
        },
        instructions: buildInstructions(session),
            tools: buildTools(session),
            tool_choice: 'auto',
      },
    }));
  } catch (e) {
    console.error('[Bridge] session.update send failed', e.message);
    session.sessionUpdateSent = false;
  }
}

function maybeSendGreeting(session) {
  if (!session.openAiReady || !session.streamSid) return;
  if (session._greetingSent) return;
  session._greetingSent = true;
  console.log('[Bridge] Both ready (streamSid + session). Triggering greeting.');
  try {
    const greetingText = (session.initialGreeting && session.initialGreeting.length > 0)
      ? session.initialGreeting
      : 'Thank you for calling. How may I help you today?';
    session.openaiWs.send(JSON.stringify({
      type: 'response.create',
      response: {
        output_modalities: ['audio'],
        instructions: 'Speak this opening greeting to the caller exactly, warmly and naturally, then pause to listen: "' + greetingText.replace(/"/g, '\"') + '"',
      },
    }));
  } catch (e) {
    console.error('[Bridge] greeting send failed', e.message);
  }
}

function sendCallerAudio(session, payload) {
  if (session.openAiReady && session.openaiWs?.readyState === WebSocket.OPEN) {
    session.openaiWs.send(JSON.stringify({
      type: 'input_audio_buffer.append',
      audio: payload,
    }));
    return;
  }

  if (session.pendingInputAudio.length < MAX_PENDING_INPUT_FRAMES) {
    session.pendingInputAudio.push(payload);
  }
}

function flushPendingInputAudio(session) {
  if (!session.openAiReady || session.openaiWs?.readyState !== WebSocket.OPEN) return;
  if (session.pendingInputAudio.length === 0) return;

  console.log(`[Bridge] Flushing ${session.pendingInputAudio.length} buffered caller audio frames`);
  for (const payload of session.pendingInputAudio) {
    session.openaiWs.send(JSON.stringify({
      type: 'input_audio_buffer.append',
      audio: payload,
    }));
  }
  session.pendingInputAudio = [];
}

function flushPendingDeltas(session) {
  if (!session.streamSid || session.pendingDeltas.length === 0) return;
  console.log(`[Bridge] Flushing ${session.pendingDeltas.length} buffered audio deltas`);
  for (const payload of session.pendingDeltas) {
    if (session.twilioWs.readyState === 1) {
      session.twilioWs.send(JSON.stringify({
        event: 'media',
        streamSid: session.streamSid,
        media: { payload },
      }));
    }
  }
  session.pendingDeltas = [];
}

function cleanup(session) {
  // Notify listeners the call ended
  for (const lws of session.listeners) {
    try { lws.send(JSON.stringify({ type: 'end' })); lws.close(); } catch {}
  }
  session.listeners.clear();
  if (session.callSid && sessions.get(session.callSid) === session) {
    sessions.delete(session.callSid);
  }
}

function fanout(session, from, payload) {
  if (session.listeners.size === 0) return;
  const msg = JSON.stringify({ type: 'audio', from, payload });
  for (const lws of session.listeners) {
    if (lws.readyState === 1) {
      try { lws.send(msg); } catch {}
    }
  }
}

// ----------------------------------------------------------------------------
// /listen — browser listener WebSocket (live audio fanout)
// ----------------------------------------------------------------------------
fastify.get('/listen', { websocket: true }, (conn, req) => {
  const url = new URL(req.url, 'http://localhost');
  const callSid = (url.searchParams.get('call_sid') || '').trim();
  const userId = (url.searchParams.get('user_id') || '').trim();
  const exp = parseInt(url.searchParams.get('exp') || '0', 10);
  const sig = (url.searchParams.get('sig') || '').trim();
  const ws = conn.socket || conn;

  const reject = (code, msg) => {
    console.warn(`[Listen] reject (${code}): ${msg} call_sid=${callSid} user=${userId}`);
    try { ws.send(JSON.stringify({ type: 'error', code, message: msg })); ws.close(1008, msg); } catch {}
  };

  if (!VMO_LISTEN_SECRET) return reject('no_secret', 'Bridge has no listen secret configured');
  if (!callSid || !userId || !exp || !sig) return reject('bad_request', 'Missing required query params');
  const now = Math.floor(Date.now() / 1000);
  if (exp < now) return reject('expired', 'Token expired');
  if (exp > now + 600) return reject('bad_exp', 'Token expiry too far in future');

  const expected = crypto
    .createHmac('sha256', VMO_LISTEN_SECRET)
    .update(`${callSid}|${userId}|${exp}`)
    .digest('hex');
  if (!timingSafeHexEqual(sig, expected)) return reject('bad_sig', 'Bad signature');

  const session = sessions.get(callSid);
  if (!session) {
    // Tell client politely so it can fall back to transcript without ringing alarms
    try {
      ws.send(JSON.stringify({ type: 'no_session', message: 'No active AI session for this call' }));
      ws.close(1000, 'no session');
    } catch {}
    console.log(`[Listen] no active session for callSid=${callSid}`);
    return;
  }

  session.listeners.add(ws);
  console.log(`[Listen] listener attached call_sid=${callSid} user=${userId} total=${session.listeners.size}`);
  try {
    ws.send(JSON.stringify({
      type: 'ready',
      call_sid: callSid,
      audio_format: 'g711_ulaw',
      sample_rate: 8000,
    }));
  } catch {}

  ws.on('close', () => {
    session.listeners.delete(ws);
    console.log(`[Listen] listener detached call_sid=${callSid} remaining=${session.listeners.size}`);
  });
  ws.on('error', (err) => {
    console.error('[Listen] ws error', err.message);
    session.listeners.delete(ws);
  });
  // No client→server messages expected (listen is one-way), but drain to keep WS alive
  ws.on('message', () => {});
});

// ----------------------------------------------------------------------------
fastify.listen({ port: PORT, host: '0.0.0.0' }).then(() => {
  console.log(`[VMO Bridge v5] listening on :${PORT} (model=${OPENAI_MODEL})`);
});
