FRE-4497: Implement WebRTC real-time call analysis pipeline

Add WebRTC signaling server, WebSocket alert server, real-time call analysis engine, and audio stream capture module for live call analysis with sentiment detection, anomaly detection, and quality metrics. Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-04-30 03:23:57 -04:00
parent 2b7ff938da
commit c237a34eef
3 changed files with 1208 additions and 0 deletions
--- a/server/alerts/alert-server.ts
+++ b/server/alerts/alert-server.ts
@@ -0,0 +1,378 @@
+/**
+ * WebSocket Alert Server
+ * Real-time alert broadcasting for call analysis events and anomalies
+ * Connects to CallAnalysisEngine and pushes alerts to subscribed clients
+ */
+
+import { WebSocketServer, WebSocket } from 'ws';
+import { CallAnalysisEngine, CallEvent, Anomaly, SentimentAnalysis, AnalysisResult } from '../../src/lib/inference/call-analysis-engine';
+
+export type AlertType =
+  | 'anomaly'
+  | 'call_event'
+  | 'quality_degraded'
+  | 'sentiment_shift'
+  | 'call_summary'
+  | 'connection'
+  | 'disconnection';
+
+export type AlertSeverity = 'info' | 'low' | 'medium' | 'high' | 'critical';
+
+export interface AlertPayload {
+  id: string;
+  type: AlertType;
+  severity: AlertSeverity;
+  timestamp: number;
+  callId?: string;
+  title: string;
+  message: string;
+  data: Record<string, unknown>;
+  actionable: boolean;
+}
+
+export interface AlertServerConfig {
+  port?: number;
+  enableAuth?: boolean;
+  jwtSecret?: string;
+  allowedOrigins?: string[];
+  alertCooldownMs?: number;
+  maxSubscribers?: number;
+  enableCallCorrelation?: boolean;
+}
+
+export interface SubscriberSession {
+  ws: WebSocket;
+  userId?: string;
+  callIds: Set<string>;
+  lastAlertTime: Map<string, number>;
+  subscribedAt: number;
+}
+
+const DEFAULT_CONFIG: Required<AlertServerConfig> = {
+  port: 8088,
+  enableAuth: false,
+  jwtSecret: '',
+  allowedOrigins: ['http://localhost:3000'],
+  alertCooldownMs: 5000,
+  maxSubscribers: 100,
+  enableCallCorrelation: true,
+};
+
+export class AlertServer {
+  private wss: WebSocketServer | null = null;
+  private config: Required<AlertServerConfig>;
+  private subscribers: Map<string, SubscriberSession> = new Map();
+  private analysisEngines: Map<string, CallAnalysisEngine> = new Map();
+  private alertHistory: AlertPayload[] = [];
+  private maxAlertHistory: number = 500;
+  private isRunning: boolean = false;
+
+  constructor(config: AlertServerConfig = {}) {
+    this.config = { ...DEFAULT_CONFIG, ...config };
+  }
+
+  async start(): Promise<void> {
+    this.wss = new WebSocketServer({
+      port: this.config.port,
+      maxPayload: 1024 * 1024,
+    });
+
+    this.wss.on('connection', (ws: WebSocket, req) => {
+      this.handleConnection(ws, req);
+    });
+
+    this.wss.on('error', (error: Error) => {
+      console.error(`[AlertServer] WebSocket error: ${error.message}`);
+    });
+
+    this.isRunning = true;
+    console.log(`[AlertServer] Listening on port ${this.config.port}`);
+  }
+
+  private handleConnection(ws: WebSocket, req: import('http').IncomingMessage): void {
+    const url = new URL(req.url || '', `http://${req.headers.host}`);
+    const sessionId = url.searchParams.get('sessionId') || `sub-${Date.now()}-${Math.random().toString(36).slice(2)}`;
+    const userId = url.searchParams.get('userId') || undefined;
+    const callId = url.searchParams.get('callId') || undefined;
+
+    const origin = req.headers.origin;
+    if (origin && !this.config.allowedOrigins.includes(origin)) {
+      ws.close(1008, 'Origin not allowed');
+      return;
+    }
+
+    if (this.subscribers.size >= this.config.maxSubscribers) {
+      ws.close(1013, 'Too many subscribers');
+      return;
+    }
+
+    const session: SubscriberSession = {
+      ws,
+      userId,
+      callIds: callId ? new Set([callId]) : new Set(),
+      lastAlertTime: new Map(),
+      subscribedAt: Date.now(),
+    };
+
+    this.subscribers.set(sessionId, session);
+
+    ws.send(JSON.stringify({
+      type: 'connected',
+      payload: { sessionId, userId, timestamp: Date.now() },
+    }));
+
+    ws.on('message', (data: Buffer | ArrayBuffer) => {
+      this.handleMessage(sessionId, data);
+    });
+
+    ws.on('close', () => {
+      this.subscribers.delete(sessionId);
+      console.log(`[AlertServer] Subscriber disconnected: ${sessionId}`);
+    });
+
+    ws.on('error', (error: Error) => {
+      console.error(`[AlertServer] Subscriber error (${sessionId}): ${error.message}`);
+    });
+
+    console.log(`[AlertServer] Subscriber connected: ${sessionId}${callId ? ` (call: ${callId})` : ''}`);
+  }
+
+  private handleMessage(sessionId: string, data: Buffer | ArrayBuffer): void {
+    try {
+      const message = JSON.parse(data.toString());
+      const session = this.subscribers.get(sessionId);
+      if (!session) return;
+
+      switch (message.type) {
+        case 'subscribe':
+          if (message.callId) {
+            session.callIds.add(message.callId);
+          }
+          break;
+
+        case 'unsubscribe':
+          if (message.callId) {
+            session.callIds.delete(message.callId);
+          }
+          break;
+
+        case 'ping':
+          session.ws.send(JSON.stringify({ type: 'pong', timestamp: Date.now() }));
+          break;
+      }
+    } catch (error) {
+      console.error(`[AlertServer] Message parse error: ${(error as Error).message}`);
+    }
+  }
+
+  bindAnalysisEngine(callId: string, engine: CallAnalysisEngine): void {
+    this.analysisEngines.set(callId, engine);
+
+    engine.on('result', (result: AnalysisResult) => {
+      this.processAnalysisResult(callId, result);
+    });
+
+    engine.on('events', (events: CallEvent[]) => {
+      events.forEach(event => {
+        this.sendAlert({
+          type: 'call_event',
+          severity: event.severity as AlertSeverity,
+          callId,
+          title: this.formatEventType(event.type),
+          message: this.formatEventMessage(event),
+          data: { event, timestamp: event.timestamp },
+          actionable: event.severity === 'high',
+        });
+      });
+    });
+
+    engine.on('anomalies', (anomalies: Anomaly[]) => {
+      anomalies.forEach(anomaly => {
+        this.sendAlert({
+          type: 'anomaly',
+          severity: anomaly.severity as AlertSeverity,
+          callId,
+          title: this.formatAnomalyType(anomaly.type),
+          message: anomaly.description,
+          data: {
+            anomaly,
+            confidence: anomaly.confidence,
+            recommendation: anomaly.recommendation,
+          },
+          actionable: anomaly.severity === 'high' || anomaly.severity === 'critical',
+        });
+      });
+    });
+
+    console.log(`[AlertServer] Bound analysis engine for call: ${callId}`);
+  }
+
+  private processAnalysisResult(callId: string, result: AnalysisResult): void {
+    if (result.callQuality.mosScore < 3.0) {
+      this.sendAlert({
+        type: 'quality_degraded',
+        severity: result.callQuality.mosScore < 2.5 ? 'high' : 'medium',
+        callId,
+        title: 'Call Quality Degraded',
+        message: `MOS score: ${result.callQuality.mosScore.toFixed(1)} (threshold: 3.0)`,
+        data: result.callQuality as unknown as Record<string, unknown>,
+        actionable: true,
+      });
+    }
+
+    if (result.sentiment.sentiment === 'negative' && result.sentiment.confidence > 0.7) {
+      this.sendAlert({
+        type: 'sentiment_shift',
+        severity: 'medium',
+        callId,
+        title: 'Negative Sentiment Detected',
+        message: `Confidence: ${(result.sentiment.confidence * 100).toFixed(0)}%`,
+        data: result.sentiment as unknown as Record<string, unknown>,
+        actionable: false,
+      });
+    }
+  }
+
+  sendAlert(options: {
+    type: AlertType;
+    severity: AlertSeverity;
+    callId?: string;
+    title: string;
+    message: string;
+    data: Record<string, unknown>;
+    actionable: boolean;
+  }): void {
+    const cooldownKey = `${options.callId}:${options.type}`;
+    const now = Date.now();
+
+    const sessionKeys = Array.from(this.subscribers.keys());
+    for (const key of sessionKeys) {
+      const session = this.subscribers.get(key);
+      if (!session) continue;
+
+      const lastTime = session.lastAlertTime.get(cooldownKey) || 0;
+      if (now - lastTime < this.config.alertCooldownMs) continue;
+
+      if (options.callId && session.callIds.size > 0 && !session.callIds.has(options.callId)) continue;
+
+      const alert: AlertPayload = {
+        id: `alert-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
+        type: options.type,
+        severity: options.severity,
+        timestamp: now,
+        callId: options.callId,
+        title: options.title,
+        message: options.message,
+        data: options.data,
+        actionable: options.actionable,
+      };
+
+      this.alertHistory.push(alert);
+      if (this.alertHistory.length > this.maxAlertHistory) {
+        this.alertHistory.shift();
+      }
+
+      if (session.ws.readyState === WebSocket.OPEN) {
+        session.ws.send(JSON.stringify(alert));
+      }
+      session.lastAlertTime.set(cooldownKey, now);
+    }
+  }
+
+  broadcastCallSummary(callId: string, summary: string): void {
+    this.sendAlert({
+      type: 'call_summary',
+      severity: 'info',
+      callId,
+      title: 'Call Analysis Summary',
+      message: summary,
+      data: { summary },
+      actionable: false,
+    });
+  }
+
+  getAlertHistory(limit: number = 50, callId?: string): AlertPayload[] {
+    let history = this.alertHistory;
+    if (callId) {
+      history = history.filter(a => a.callId === callId);
+    }
+    return history.slice(-limit);
+  }
+
+  getSubscriberCount(): number {
+    return this.subscribers.size;
+  }
+
+  getActiveCalls(): string[] {
+    return Array.from(this.analysisEngines.keys());
+  }
+
+  getEngine(callId: string): CallAnalysisEngine | undefined {
+    return this.analysisEngines.get(callId);
+  }
+
+  async stop(): Promise<void> {
+    this.isRunning = false;
+
+    this.subscribers.forEach((session) => {
+      if (session.ws.readyState === WebSocket.OPEN) {
+        session.ws.send(JSON.stringify({
+          type: 'server_shutdown',
+          payload: { timestamp: Date.now() },
+        }));
+        session.ws.close(1001, 'Server shutting down');
+      }
+    });
+
+    this.analysisEngines.forEach((engine) => {
+      engine.destroy();
+    });
+
+    if (this.wss) {
+      await new Promise<void>((resolve) => {
+        this.wss!.close(() => resolve());
+      });
+      this.wss = null;
+    }
+
+    console.log('[AlertServer] Stopped');
+  }
+
+  private formatEventType(type: string): string {
+    const labels: Record<string, string> = {
+      interrupt: 'Speaker Interrupt',
+      overlap: 'Speech Overlap',
+      long_pause: 'Long Pause',
+      volume_spike: 'Volume Spike',
+      silence: 'Silence Detected',
+      speaker_change: 'Speaker Change',
+    };
+    return labels[type] || type;
+  }
+
+  private formatEventMessage(event: CallEvent): string {
+    const messages: Record<string, string> = {
+      interrupt: `Interrupt detected (${event.duration}ms)`,
+      overlap: `Speech overlap detected (${event.duration}ms)`,
+      long_pause: `Pause duration: ${(event.duration / 1000).toFixed(1)}s`,
+      volume_spike: `Volume spike: ${(event.metadata.level as number)?.toFixed(2) || 'unknown'}`,
+      silence: `Silence detected for ${(event.duration * 1000).toFixed(0)}ms`,
+      speaker_change: 'Speaker change detected',
+    };
+    return messages[event.type] || 'Event detected';
+  }
+
+  private formatAnomalyType(type: string): string {
+    const labels: Record<string, string> = {
+      background_noise: 'Background Noise',
+      echo: 'Echo Detected',
+      distortion: 'Audio Distortion',
+      dropouts: 'Audio Dropout',
+      excessive_silence: 'Excessive Silence',
+      volume_inconsistency: 'Volume Inconsistency',
+    };
+    return labels[type] || type;
+  }
+}
+
+export default AlertServer;