feat(whatsapp-call): add server-side WebRTC media server for call persistence
Implements a Pion Go B2BUA media server sidecar that sits between Meta's WhatsApp Cloud API and the agent's browser, enabling call persistence across page reloads, server-side recording, multi-participant support, audio injection, and AI integration readiness. Go Media Server (enterprise/media-server/): - Pion WebRTC v4 B2BUA with Peer A (Meta) and Peer B (Agent) connections - Real-time OGG/Opus recording with crash recovery - Audio bridge with multi-peer fan-out and AudioConsumer plugin interface - Audio injection from OGG files with loop support for hold music - Session manager with graceful shutdown and orphaned recording recovery - HTTP API with Bearer token auth, health checks, and metrics Rails Integration: - Whatsapp::MediaServerClient HTTP client for Go sidecar communication - Dual-mode CallService: legacy browser-direct and server-relay paths - Media server callback controller for agent disconnect/recording/terminate - CallRecordingFetchJob: downloads OGG from Go server to ActiveStorage/S3 - CallCleanupJob: sweeps stale ringing and in-progress calls - New endpoints: active, agent_answer, reconnect, join, play_audio - DB migration: media_session_id column with indexes Frontend: - Dual-mode composable auto-detecting legacy vs server-relay - handleAgentOffer() for receiving SDP from media server - useCallReconnection composable for page reload recovery - Removed terminateCallOnUnload in server-relay mode - Simplified outbound call flow in ConversationHeader - New ActionCable event: whatsapp_call.agent_offer Documentation: - Server-side WebRTC architecture spec (1505 lines) - Implementation plan with 119 trackable checklist items - Feature spec, PR breakdown, and relay architecture docs
This commit is contained in:
tds-1
@@ -3,18 +3,18 @@ import { useI18n } from 'vue-i18n';
|
||||
import {
|
||||
useWhatsappCallsStore,
|
||||
getOutboundCallState,
|
||||
cleanupOutboundCall,
|
||||
} from 'dashboard/stores/whatsappCalls';
|
||||
import WhatsappCallsAPI from 'dashboard/api/whatsappCalls';
|
||||
import Auth from 'dashboard/api/auth';
|
||||
import Timer from 'dashboard/helper/Timer';
|
||||
|
||||
// ── Module-level WebRTC state for inbound calls accepted from anywhere ──
|
||||
// Kept at module scope so both the composable and acceptWhatsappCallById share it.
|
||||
// ── Module-level WebRTC state (shared across legacy inbound + server-relay) ──
|
||||
let inboundPc = null;
|
||||
let inboundStream = null;
|
||||
let inboundAudio = null;
|
||||
|
||||
// ── Module-level recording state ──
|
||||
// ── Module-level recording state (legacy mode only) ──
|
||||
let mediaRecorder = null;
|
||||
let recordedChunks = [];
|
||||
let recordingCallId = null;
|
||||
@@ -40,19 +40,18 @@ function cleanupInboundWebRTC() {
|
||||
/**
|
||||
* Start recording both local and remote audio tracks via MediaRecorder.
|
||||
* Mixes them into a single stream using AudioContext.
|
||||
* Used ONLY in legacy (browser-direct) mode.
|
||||
*/
|
||||
export function startCallRecording(pc, localStream, callId) {
|
||||
try {
|
||||
const ctx = new AudioContext();
|
||||
const dest = ctx.createMediaStreamDestination();
|
||||
|
||||
// Add local mic track
|
||||
if (localStream) {
|
||||
const localSource = ctx.createMediaStreamSource(localStream);
|
||||
localSource.connect(dest);
|
||||
}
|
||||
|
||||
// Add remote tracks from peer connection
|
||||
pc.getReceivers().forEach(receiver => {
|
||||
if (receiver.track && receiver.track.kind === 'audio') {
|
||||
const remoteStream = new MediaStream([receiver.track]);
|
||||
@@ -81,6 +80,7 @@ export function startCallRecording(pc, localStream, callId) {
|
||||
|
||||
/**
|
||||
* Stop recording and upload the audio blob to the backend.
|
||||
* Used ONLY in legacy (browser-direct) mode.
|
||||
*/
|
||||
function stopAndUploadRecording(callId) {
|
||||
if (!mediaRecorder || mediaRecorder.state === 'inactive') return;
|
||||
@@ -110,7 +110,17 @@ function waitForIceGatheringComplete(pc) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
const timeout = setTimeout(() => {
|
||||
|
||||
let timeout = null;
|
||||
|
||||
const cleanup = () => {
|
||||
clearTimeout(timeout);
|
||||
pc.onicegatheringstatechange = null;
|
||||
pc.oniceconnectionstatechange = null;
|
||||
};
|
||||
|
||||
timeout = setTimeout(() => {
|
||||
cleanup();
|
||||
// eslint-disable-next-line no-console
|
||||
console.warn(
|
||||
'[WhatsApp Call] ICE gathering timed out, sending partial SDP'
|
||||
@@ -120,85 +130,155 @@ function waitForIceGatheringComplete(pc) {
|
||||
|
||||
pc.onicegatheringstatechange = () => {
|
||||
if (pc.iceGatheringState === 'complete') {
|
||||
clearTimeout(timeout);
|
||||
cleanup();
|
||||
resolve();
|
||||
}
|
||||
};
|
||||
pc.oniceconnectionstatechange = () => {
|
||||
if (pc.iceConnectionState === 'failed') {
|
||||
clearTimeout(timeout);
|
||||
cleanup();
|
||||
reject(new Error('ICE connection failed'));
|
||||
}
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
// ── Server-relay mode: detect by absence of sdpOffer in incoming call data ──
|
||||
function isServerRelayCall(call) {
|
||||
return !call.sdpOffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* Core accept logic: creates WebRTC session and posts SDP to backend.
|
||||
* Can be called from anywhere — composable, widget, or bubble.
|
||||
* Returns { success: true } or { success: false, error }.
|
||||
* Handle an SDP offer from the media server (Peer B). Used in server-relay mode
|
||||
* for both inbound accept and outbound connect flows.
|
||||
*
|
||||
* Flow: getUserMedia -> RTCPeerConnection(iceServers) -> setRemoteDescription(offer)
|
||||
* -> createAnswer -> waitForICE -> POST /agent_answer
|
||||
*/
|
||||
async function doAcceptCall(call) {
|
||||
async function handleAgentOffer(callId, sdpOffer, iceServers) {
|
||||
cleanupInboundWebRTC();
|
||||
|
||||
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
|
||||
inboundStream = stream;
|
||||
try {
|
||||
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
|
||||
inboundStream = stream;
|
||||
|
||||
const iceServers = call.iceServers?.length
|
||||
? call.iceServers
|
||||
: [{ urls: 'stun:stun.l.google.com:19302' }];
|
||||
const servers = iceServers?.length
|
||||
? iceServers
|
||||
: [{ urls: 'stun:stun.l.google.com:19302' }];
|
||||
|
||||
const pc = new RTCPeerConnection({ iceServers });
|
||||
inboundPc = pc;
|
||||
const pc = new RTCPeerConnection({ iceServers: servers });
|
||||
inboundPc = pc;
|
||||
|
||||
stream.getTracks().forEach(track => pc.addTrack(track, stream));
|
||||
stream.getTracks().forEach(track => pc.addTrack(track, stream));
|
||||
|
||||
pc.ontrack = event => {
|
||||
const [remoteStream] = event.streams;
|
||||
if (!remoteStream) return;
|
||||
if (!inboundAudio) {
|
||||
const audio = document.createElement('audio');
|
||||
audio.autoplay = true;
|
||||
document.body.appendChild(audio);
|
||||
inboundAudio = audio;
|
||||
}
|
||||
inboundAudio.srcObject = remoteStream;
|
||||
inboundAudio.play().catch(() => {});
|
||||
pc.ontrack = event => {
|
||||
const [remoteStream] = event.streams;
|
||||
if (!remoteStream) return;
|
||||
if (!inboundAudio) {
|
||||
const audio = document.createElement('audio');
|
||||
audio.autoplay = true;
|
||||
document.body.appendChild(audio);
|
||||
inboundAudio = audio;
|
||||
}
|
||||
inboundAudio.srcObject = remoteStream;
|
||||
inboundAudio.play().catch(() => {});
|
||||
|
||||
// Start recording once remote audio is available
|
||||
startCallRecording(pc, stream, call.id);
|
||||
};
|
||||
// No client-side recording in server-relay mode — the media server records
|
||||
};
|
||||
|
||||
await pc.setRemoteDescription({ type: 'offer', sdp: call.sdpOffer });
|
||||
const answer = await pc.createAnswer();
|
||||
await pc.setLocalDescription(answer);
|
||||
await waitForIceGatheringComplete(pc);
|
||||
await pc.setRemoteDescription({ type: 'offer', sdp: sdpOffer });
|
||||
const answer = await pc.createAnswer();
|
||||
await pc.setLocalDescription(answer);
|
||||
await waitForIceGatheringComplete(pc);
|
||||
|
||||
const completeSdp = pc.localDescription.sdp;
|
||||
await WhatsappCallsAPI.accept(call.id, completeSdp);
|
||||
const completeSdp = pc.localDescription.sdp;
|
||||
await WhatsappCallsAPI.agentAnswer(callId, completeSdp);
|
||||
|
||||
return { success: true };
|
||||
return { success: true };
|
||||
} catch (err) {
|
||||
cleanupInboundWebRTC();
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// Expose handleAgentOffer so ActionCable handler can invoke it
|
||||
export { handleAgentOffer };
|
||||
|
||||
/**
|
||||
* Legacy mode: creates WebRTC session and posts SDP to backend (browser ↔ Meta).
|
||||
* Can be called from anywhere — composable, widget, or bubble.
|
||||
*/
|
||||
async function doAcceptCall(call) {
|
||||
// Server-relay mode: just POST /accept without SDP. Wait for agent_offer event.
|
||||
if (isServerRelayCall(call)) {
|
||||
await WhatsappCallsAPI.accept(call.id);
|
||||
return { success: true, awaitingAgentOffer: true };
|
||||
}
|
||||
|
||||
// Legacy mode: full browser-side WebRTC handshake
|
||||
cleanupInboundWebRTC();
|
||||
|
||||
try {
|
||||
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
|
||||
inboundStream = stream;
|
||||
|
||||
const iceServers = call.iceServers?.length
|
||||
? call.iceServers
|
||||
: [{ urls: 'stun:stun.l.google.com:19302' }];
|
||||
|
||||
const pc = new RTCPeerConnection({ iceServers });
|
||||
inboundPc = pc;
|
||||
|
||||
stream.getTracks().forEach(track => pc.addTrack(track, stream));
|
||||
|
||||
pc.ontrack = event => {
|
||||
const [remoteStream] = event.streams;
|
||||
if (!remoteStream) return;
|
||||
if (!inboundAudio) {
|
||||
const audio = document.createElement('audio');
|
||||
audio.autoplay = true;
|
||||
document.body.appendChild(audio);
|
||||
inboundAudio = audio;
|
||||
}
|
||||
inboundAudio.srcObject = remoteStream;
|
||||
inboundAudio.play().catch(() => {});
|
||||
|
||||
// Start recording once remote audio is available (legacy mode only)
|
||||
startCallRecording(pc, stream, call.id);
|
||||
};
|
||||
|
||||
await pc.setRemoteDescription({ type: 'offer', sdp: call.sdpOffer });
|
||||
const answer = await pc.createAnswer();
|
||||
await pc.setLocalDescription(answer);
|
||||
await waitForIceGatheringComplete(pc);
|
||||
|
||||
const completeSdp = pc.localDescription.sdp;
|
||||
await WhatsappCallsAPI.accept(call.id, completeSdp);
|
||||
|
||||
return { success: true };
|
||||
} catch (err) {
|
||||
cleanupInboundWebRTC();
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Standalone function callable from VoiceCall bubble.
|
||||
* Fetches call data if needed, runs WebRTC accept, updates store.
|
||||
*/
|
||||
export async function acceptWhatsappCallById(waCallId) {
|
||||
export async function acceptWhatsappCallById(callId) {
|
||||
const callsStore = useWhatsappCallsStore();
|
||||
|
||||
if (callsStore.hasActiveCall) {
|
||||
return { success: false, error: 'active_call_exists' };
|
||||
}
|
||||
|
||||
// 1. Check if the call is already in the incoming store
|
||||
let call = callsStore.incomingCalls.find(
|
||||
c => c.id === waCallId || c.callId === String(waCallId)
|
||||
c => c.id === callId || c.callId === String(callId)
|
||||
);
|
||||
|
||||
// 2. Not in store (page was refreshed) → fetch from API
|
||||
if (!call) {
|
||||
const { data } = await WhatsappCallsAPI.show(waCallId);
|
||||
const { data } = await WhatsappCallsAPI.show(callId);
|
||||
if (data.status !== 'ringing') {
|
||||
return { success: false, error: 'not_ringing' };
|
||||
}
|
||||
@@ -215,19 +295,30 @@ export async function acceptWhatsappCallById(waCallId) {
|
||||
callsStore.addIncomingCall(call);
|
||||
}
|
||||
|
||||
// 3. Run the WebRTC accept
|
||||
await doAcceptCall(call);
|
||||
try {
|
||||
const result = await doAcceptCall(call);
|
||||
|
||||
// 4. Move from incoming to active
|
||||
callsStore.removeIncomingCall(call.callId);
|
||||
callsStore.setActiveCall({ ...call });
|
||||
callsStore.removeIncomingCall(call.callId);
|
||||
|
||||
return { success: true, call };
|
||||
// In server-relay mode the call becomes active but awaits the agent_offer
|
||||
// ActionCable event to complete WebRTC setup. Mark it with serverRelay flag.
|
||||
const activeCallData = {
|
||||
...call,
|
||||
serverRelay: isServerRelayCall(call),
|
||||
};
|
||||
callsStore.setActiveCall(activeCallData);
|
||||
|
||||
return { success: true, call: activeCallData, ...result };
|
||||
} catch (err) {
|
||||
callsStore.removeIncomingCall(call.callId);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire-and-forget terminate request using fetch + keepalive.
|
||||
* Works reliably inside beforeunload / pagehide where axios won't complete.
|
||||
* Used ONLY in legacy mode. Server-relay mode does NOT terminate on unload.
|
||||
*/
|
||||
function terminateCallOnUnload(callId) {
|
||||
const authData = Auth.hasAuthCookie() ? Auth.getAuthData() : {};
|
||||
@@ -260,9 +351,10 @@ export function useWhatsappCallSession() {
|
||||
const isMuted = ref(false);
|
||||
const callError = ref(null);
|
||||
const callDuration = ref(0);
|
||||
const isReconnecting = computed(() => callsStore.isReconnecting);
|
||||
|
||||
const durationTimer = new Timer(elapsed => {
|
||||
callDuration.value = elapsed;
|
||||
callDuration.value = callsStore.callTimerOffset + elapsed;
|
||||
});
|
||||
|
||||
const activeCall = computed(() => callsStore.activeCall);
|
||||
@@ -285,16 +377,27 @@ export function useWhatsappCallSession() {
|
||||
|
||||
// Register cleanup so external call-end events can teardown WebRTC
|
||||
callsStore.registerCleanupCallback(() => {
|
||||
stopAndUploadRecording();
|
||||
// Only do recording cleanup in legacy mode
|
||||
if (!callsStore.isMediaServerEnabled) {
|
||||
stopAndUploadRecording();
|
||||
}
|
||||
cleanupInboundWebRTC();
|
||||
durationTimer.stop();
|
||||
callDuration.value = 0;
|
||||
});
|
||||
|
||||
// Terminate active call on page close / reload
|
||||
// On page close / reload:
|
||||
// - Legacy mode: terminate call (current behavior)
|
||||
// - Server-relay mode: just clean up local WebRTC resources, call persists
|
||||
const handleBeforeUnload = () => {
|
||||
const call = callsStore.activeCall;
|
||||
if (call?.id) {
|
||||
if (!call?.id) return;
|
||||
|
||||
if (call.serverRelay) {
|
||||
// Server-relay: only clean up local resources, do NOT terminate
|
||||
cleanupInboundWebRTC();
|
||||
} else {
|
||||
// Legacy: terminate and clean up
|
||||
terminateCallOnUnload(call.id);
|
||||
cleanupInboundWebRTC();
|
||||
}
|
||||
@@ -314,7 +417,6 @@ export function useWhatsappCallSession() {
|
||||
|
||||
/**
|
||||
* Accept an incoming call — used by the floating widget buttons.
|
||||
* Uses the same doAcceptCall core + starts the timer.
|
||||
*/
|
||||
const acceptCall = async call => {
|
||||
if (isAccepting.value) return;
|
||||
@@ -322,10 +424,21 @@ export function useWhatsappCallSession() {
|
||||
callError.value = null;
|
||||
|
||||
try {
|
||||
await doAcceptCall(call);
|
||||
const result = await doAcceptCall(call);
|
||||
callsStore.removeIncomingCall(call.callId);
|
||||
callsStore.setActiveCall({ ...call });
|
||||
durationTimer.start();
|
||||
|
||||
const activeCallData = {
|
||||
...call,
|
||||
serverRelay: isServerRelayCall(call),
|
||||
};
|
||||
callsStore.setActiveCall(activeCallData);
|
||||
|
||||
// In legacy mode, WebRTC is already established so start timer now.
|
||||
// In server-relay mode, timer starts when handleAgentOffer completes
|
||||
// (triggered by the whatsapp_call.agent_offer ActionCable event).
|
||||
if (!result.awaitingAgentOffer) {
|
||||
durationTimer.start();
|
||||
}
|
||||
} catch (err) {
|
||||
callError.value =
|
||||
err.name === 'NotAllowedError'
|
||||
@@ -333,7 +446,7 @@ export function useWhatsappCallSession() {
|
||||
: t('WHATSAPP_CALL.CALL_FAILED');
|
||||
// eslint-disable-next-line no-console
|
||||
console.error('[WhatsApp Call] acceptCall error:', err);
|
||||
cleanupInboundWebRTC();
|
||||
// Note: doAcceptCall already cleans up WebRTC resources on error
|
||||
} finally {
|
||||
isAccepting.value = false;
|
||||
}
|
||||
@@ -353,7 +466,10 @@ export function useWhatsappCallSession() {
|
||||
const call = activeCall.value;
|
||||
if (!call) return;
|
||||
|
||||
stopAndUploadRecording(call.id);
|
||||
// Only upload recording in legacy mode
|
||||
if (!call.serverRelay) {
|
||||
stopAndUploadRecording(call.id);
|
||||
}
|
||||
|
||||
try {
|
||||
await WhatsappCallsAPI.terminate(call.id);
|
||||
@@ -361,7 +477,10 @@ export function useWhatsappCallSession() {
|
||||
// Best effort
|
||||
} finally {
|
||||
cleanupInboundWebRTC();
|
||||
callsStore.handleCallEnded(call.callId);
|
||||
cleanupOutboundCall();
|
||||
// Clear state directly — do NOT use handleCallEnded here since that is
|
||||
// meant for external events (ActionCable) and would invoke cleanupCallback
|
||||
// which would duplicate the cleanup we just performed.
|
||||
callsStore.clearActiveCall();
|
||||
durationTimer.stop();
|
||||
callDuration.value = 0;
|
||||
@@ -381,6 +500,14 @@ export function useWhatsappCallSession() {
|
||||
callsStore.removeIncomingCall(call.callId);
|
||||
};
|
||||
|
||||
/**
|
||||
* Start the duration timer. Called externally after server-relay WebRTC
|
||||
* setup completes (handleAgentOffer).
|
||||
*/
|
||||
const startDurationTimer = () => {
|
||||
durationTimer.start();
|
||||
};
|
||||
|
||||
onUnmounted(() => {
|
||||
window.removeEventListener('beforeunload', handleBeforeUnload);
|
||||
durationTimer.stop();
|
||||
@@ -395,6 +522,7 @@ export function useWhatsappCallSession() {
|
||||
isAccepting,
|
||||
isMuted,
|
||||
isOutboundRinging,
|
||||
isReconnecting,
|
||||
callError,
|
||||
formattedCallDuration,
|
||||
acceptCall,
|
||||
@@ -402,5 +530,6 @@ export function useWhatsappCallSession() {
|
||||
endActiveCall,
|
||||
toggleMute,
|
||||
dismissIncomingCall,
|
||||
startDurationTimer,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user