Software: Apache. PHP/8.1.30 

uname -a: Linux server1.tuhinhossain.com 5.15.0-163-generic #173-Ubuntu SMP Tue Oct 14 17:51:00 UTC
2025 x86_64 

uid=1002(picotech) gid=1003(picotech) groups=1003(picotech),0(root)  

Safe-mode: OFF (not secure)

/home/picotech/domains/note.picotech.app/public_html/src/services/   drwxr-xr-x
Free 25.88 GB of 117.98 GB (21.93%)
                            Encoder    Tools    Proc.    FTP brute    Sec.    SQL    PHP-code    Update    Self remove    Logout    

Viewing file:     liveTranscriptionService.js (10.9 KB)      -rw-r--r--

const WebSocket = require('ws');
const jwt = require('jsonwebtoken');
const config = require('../config/config');
const { Meeting } = require('../models');

let globalServiceInstance = null;

const createLiveTranscriptionService = (server) => {
  const wss = new WebSocket.Server({
    server,
    path: '/socket',
    perMessageDeflate: false,
  });

  const activeSessions = new Map();

  wss.on('connection', handleConnection);
  console.log('🔴 Live transcription WebSocket server initialized');

  async function handleConnection(ws, request) {
    try {
      // Extract token from query parameters
      const url = require('url').parse(request.url, true);
      const token = url.query.token;
      const meetingIdRaw = url.query.meetingId;
      const meetingId = parseInt(meetingIdRaw, 10);

      if (!token || !meetingIdRaw || isNaN(meetingId) ||
          token === 'undefined' || meetingIdRaw === 'undefined' ||
          token === 'null' || meetingIdRaw === 'null' ||
          token === '' || meetingIdRaw === '') {
        console.log('❌ Validation failed: Missing or invalid token/meetingId');
        ws.send(JSON.stringify({
          type: 'error',
          payload: { message: 'Missing authentication token or meeting ID' }
        }));
        ws.close(1008, 'Missing authentication token or meeting ID');
        return;
      }

      // Verify JWT token
      let decoded;
      try {
        decoded = jwt.verify(token, config.jwt.secret);
      } catch (jwtError) {
        console.error('❌ JWT verification failed:', jwtError.message);
        ws.send(JSON.stringify({
          type: 'error',
          payload: { message: 'Invalid authentication token' }
        }));
        ws.close(1008, 'Invalid authentication token');
        return;
      }
      const userId = decoded.userId;

      // Additional security: Check token expiration and validate user exists
      const { User } = require('../models');
      const user = await User.findByPk(userId);
      if (!user || !user.is_active) {
        console.error('❌ User not found or inactive');
        ws.send(JSON.stringify({
          type: 'error',
          payload: { message: 'User not found or inactive' }
        }));
        ws.close(1008, 'User not found or inactive');
        return;
      }

      // Verify meeting access
      const meeting = await Meeting.findOne({
        where: { id: meetingId, user_id: userId }
      });

      if (!meeting) {
        ws.send(JSON.stringify({
          type: 'error',
          payload: { message: 'Meeting not found or access denied' }
        }));
        ws.close(1008, 'Meeting not found or access denied');
        return;
      }

      const sessionId = `meeting_${meetingId}_${Date.now()}`;
      const session = {
        id: sessionId,
        meetingId,
        userId,
        ws,
        startTime: new Date(),
        isActive: true,
        lastActivity: Date.now(),
        connectionTimeout: setTimeout(() => {
          console.log(`Connection timeout for session ${sessionId}`);
          ws.close(1008, 'Connection timeout');
        }, 24 * 60 * 60 * 1000) // 24 hours
      };

      activeSessions.set(sessionId, session);

      // Send connection confirmation
      ws.send(JSON.stringify({
        type: 'connected',
        sessionId,
        message: 'Connected to live transcription service'
      }));

      // Set up event handlers
      ws.on('message', (data) => handleMessage(sessionId, data));
      ws.on('close', () => handleDisconnection(sessionId));
      ws.on('error', (error) => handleError(sessionId, error));

      // Start heartbeat
      startHeartbeat(sessionId);

      console.log(`🔴 Client connected to live transcription: ${sessionId}`);

    } catch (error) {
      console.error('WebSocket connection error:', error);
      ws.send(JSON.stringify({
        type: 'error',
        payload: { message: 'Authentication failed' }
      }));
      ws.close(1008, 'Authentication failed');
    }
  }

  async function handleMessage(sessionId, data) {
    const session = activeSessions.get(sessionId);
    if (!session) return;

    try {
      // Enhanced rate limiting: Check message frequency and connection duration
      const now = Date.now();
      const connectionDuration = now - session.connectionStartTime;

      // Stricter rate limiting for new connections (first 30 seconds)
      const isNewConnection = connectionDuration < 30000;
      const maxMessagesPerSecond = isNewConnection ? 10 : 50;
      const minMessageInterval = isNewConnection ? 200 : 100; // 200ms for new, 100ms for established

      if (now - session.lastMessageTime < minMessageInterval) {
        session.messageCount++;
        if (session.messageCount > maxMessagesPerSecond) {
          console.warn(`Rate limit exceeded for session ${sessionId} (${isNewConnection ? 'new' : 'established'} connection)`);
          session.ws.close(1008, 'Rate limit exceeded');
          return;
        }
      } else {
        session.messageCount = 1;
        session.lastMessageTime = now;
      }

      // Additional check: Max 1000 messages per connection
      session.totalMessageCount = (session.totalMessageCount || 0) + 1;
      if (session.totalMessageCount > 1000) {
        console.warn(`Message limit exceeded for session ${sessionId}`);
        session.ws.close(1008, 'Message limit exceeded');
        return;
      }

      const message = JSON.parse(data.toString());

      switch (message.type) {
        case 'ping':
          session.ws.send(JSON.stringify({ type: 'pong' }));
          break;

        case 'generate_summary':
          handleGenerateSummary(sessionId, message.payload);
          break;

        default:
          console.warn(`Unknown message type: ${message.type}`);
          session.ws.send(JSON.stringify({
            type: 'error',
            payload: { message: 'Unknown message type' }
          }));
      }

      session.lastActivity = Date.now();
    } catch (error) {
      console.error('Error handling message:', error);
      session.ws.send(JSON.stringify({
        type: 'error',
        payload: { message: 'Invalid message format' }
      }));
    }
  }

  function handleDisconnection(sessionId) {
    const session = activeSessions.get(sessionId);
    if (session) {
      session.isActive = false;

      // Clear connection timeout
      if (session.connectionTimeout) {
        clearTimeout(session.connectionTimeout);
      }

      // Clear heartbeat
      if (session.heartbeat) {
        clearInterval(session.heartbeat);
      }

      // Clean up
      activeSessions.delete(sessionId);

      console.log(`🔌 Client disconnected from live transcription: ${sessionId}`);
    }
  }

  function handleError(sessionId, error) {
    console.error(`WebSocket error for session ${sessionId}:`, error);
    handleDisconnection(sessionId);
  }

  function startHeartbeat(sessionId) {
    const session = activeSessions.get(sessionId);
    if (!session) return;

    const heartbeat = setInterval(() => {
      if (session.ws.readyState === WebSocket.OPEN) {
        session.ws.send(JSON.stringify({ type: 'ping' }));
      } else {
        clearInterval(heartbeat);
      }
    }, 30000); // 30 seconds

    // Store heartbeat reference for cleanup
    session.heartbeat = heartbeat;

    // Set up connection rate limiting with enhanced security
    session.messageCount = 0;
    session.lastMessageTime = Date.now();
    session.connectionStartTime = Date.now();

    // Implement connection timeout (max 2 hours per session)
    session.connectionTimeout = setTimeout(() => {
      console.log(`Connection timeout for session ${sessionId} (2 hours)`);
      session.ws.close(1008, 'Connection timeout - maximum session duration exceeded');
    }, 2 * 60 * 60 * 1000); // 2 hours
  }

  // Broadcast transcription update to all clients in a meeting
  function broadcastTranscriptionUpdate(meetingId, transcriptionData) {
    const message = {
      type: 'transcription_update',
      payload: transcriptionData,
      timestamp: new Date().toISOString()
    };
    console.log(`Broadcasting data:`, message);
    console.log(`Active sessions count:`, activeSessions.size);
    console.log(`Target meetingId:`, meetingId, `(${typeof meetingId})`);

    // Check if we have the global instance
    console.log(`Global service instance exists:`, !!globalServiceInstance);

    activeSessions.forEach((session, sessionId) => {
      console.log(`Session ${sessionId}: meetingId=${session.meetingId} (${typeof session.meetingId}), isActive=${session.isActive}, ws.readyState=${session.ws.readyState}`);
      if (session.meetingId == meetingId && session.ws.readyState === WebSocket.OPEN) {
        try {
          session.ws.send(JSON.stringify(message));
          console.log(`Sent transcription update to session ${sessionId}`);
        } catch (error) {
          console.error(`Failed to send transcription update to session ${sessionId}:`, error);
        }
      }
    });
  }

  // Broadcast summary update to all clients in a meeting
  function broadcastSummaryUpdate(meetingId, summaryData) {
    const message = {
      type: 'summary_update',
      payload: summaryData,
      timestamp: new Date().toISOString()
    };
    console.log(`Broadcasting summary data:`, message);

    activeSessions.forEach((session, sessionId) => {
      if (session.meetingId == meetingId && session.ws.readyState === WebSocket.OPEN) {
        try {
          session.ws.send(JSON.stringify(message));
          console.log(`Sent summary update to session ${sessionId}`);
        } catch (error) {
          console.error(`Failed to send summary update to session ${sessionId}:`, error);
        }
      }
    });
  }

  // Handle summary generation request
  async function handleGenerateSummary(sessionId, payload) {
    const session = activeSessions.get(sessionId);
    if (!session) return;

    try {
      console.log(`Starting summary generation for meeting ${session.meetingId}`);

      // Import the streaming summary service
      const { generateStreamingSummary } = require('./streamingSummaryService');

      // Start streaming summary generation
      await generateStreamingSummary(session.meetingId, (update) => {
        broadcastSummaryUpdate(session.meetingId, update);
      });

      console.log(`Summary generation completed for meeting ${session.meetingId}`);
    } catch (error) {
      console.error(`Error generating summary for session ${sessionId}:`, error);
      session.ws.send(JSON.stringify({
        type: 'error',
        payload: { message: 'Failed to generate summary', error: error.message }
      }));
    }
  }

  // Cleanup method
  function cleanup() {
    wss.close();
    activeSessions.clear();
  }

  // Store the service instance globally for access from other modules
  globalServiceInstance = {
    cleanup,
    broadcastTranscriptionUpdate,
    broadcastSummaryUpdate
  };

  return globalServiceInstance;
};

module.exports = createLiveTranscriptionService;

// Export the global instance for direct access
module.exports.getInstance = () => globalServiceInstance;