import { WebSocketGateway, SubscribeMessage, MessageBody, ConnectedSocket, OnGatewayDisconnect, } from '@nestjs/websockets'; import { Logger } from '@nestjs/common'; import { Socket } from 'socket.io'; import WebSocket from 'ws'; import { CallAssistService } from './call-assist.service'; type SessionState = { deepgramWs: WebSocket | null; transcript: string; context: string; suggestionTimer: NodeJS.Timeout | null; }; @WebSocketGateway({ cors: { origin: process.env.CORS_ORIGIN ?? '*', credentials: true }, namespace: '/call-assist', }) export class CallAssistGateway implements OnGatewayDisconnect { private readonly logger = new Logger(CallAssistGateway.name); private readonly sessions = new Map(); private readonly deepgramApiKey: string; constructor(private readonly callAssist: CallAssistService) { this.deepgramApiKey = process.env.DEEPGRAM_API_KEY ?? ''; } @SubscribeMessage('call-assist:start') async handleStart( @ConnectedSocket() client: Socket, @MessageBody() data: { ucid: string; leadId?: string; callerPhone?: string }, ) { this.logger.log(`Call assist start: ucid=${data.ucid} lead=${data.leadId ?? 'none'}`); const context = await this.callAssist.loadCallContext( data.leadId ?? null, data.callerPhone ?? null, ); client.emit('call-assist:context', { context: context.substring(0, 200) + '...' }); const session: SessionState = { deepgramWs: null, transcript: '', context, suggestionTimer: null, }; if (this.deepgramApiKey) { const dgUrl = `wss://api.deepgram.com/v1/listen?model=nova-2&language=en&smart_format=true&interim_results=true&endpointing=300&sample_rate=16000&encoding=linear16&channels=1`; const dgWs = new WebSocket(dgUrl, { headers: { Authorization: `Token ${this.deepgramApiKey}` }, }); dgWs.on('open', () => { this.logger.log(`Deepgram connected for ${data.ucid}`); }); dgWs.on('message', (raw: WebSocket.Data) => { try { const result = JSON.parse(raw.toString()); const text = result.channel?.alternatives?.[0]?.transcript; if (!text) return; const isFinal = result.is_final; client.emit('call-assist:transcript', { text, isFinal }); if (isFinal) { session.transcript += `Customer: ${text}\n`; } } catch {} }); dgWs.on('error', (err) => { this.logger.error(`Deepgram error: ${err.message}`); }); dgWs.on('close', () => { this.logger.log(`Deepgram closed for ${data.ucid}`); }); session.deepgramWs = dgWs; } else { this.logger.warn('DEEPGRAM_API_KEY not set — transcription disabled'); client.emit('call-assist:error', { message: 'Transcription not configured' }); } // AI suggestion every 10 seconds session.suggestionTimer = setInterval(async () => { if (!session.transcript.trim()) return; const suggestion = await this.callAssist.getSuggestion(session.transcript, session.context); if (suggestion) { client.emit('call-assist:suggestion', { text: suggestion }); } }, 10000); this.sessions.set(client.id, session); } @SubscribeMessage('call-assist:audio') handleAudio( @ConnectedSocket() client: Socket, @MessageBody() audioData: ArrayBuffer, ) { const session = this.sessions.get(client.id); if (session?.deepgramWs?.readyState === WebSocket.OPEN) { session.deepgramWs.send(Buffer.from(audioData)); } } @SubscribeMessage('call-assist:stop') handleStop(@ConnectedSocket() client: Socket) { this.cleanup(client.id); this.logger.log(`Call assist stopped: ${client.id}`); } handleDisconnect(client: Socket) { this.cleanup(client.id); } private cleanup(clientId: string) { const session = this.sessions.get(clientId); if (session) { if (session.suggestionTimer) clearInterval(session.suggestionTimer); if (session.deepgramWs) { try { session.deepgramWs.close(); } catch {} } this.sessions.delete(clientId); } } }