Stream Pumpfun Chat Messages

pumpfun-livestream-chat-websocket.js

/**
 * Custom Reverse-Engineered Pump Chat Client
 * 
 * This is a custom implementation based on reverse engineering the pump-chat-client
 * library. It demonstrates understanding of the Socket.IO protocol used by pump.fun
 * for livestream chat communication.
 * 
 * Key Features:
 * - WebSocket connection to pump.fun chat servers
 * - Socket.IO protocol implementation
 * - Event-driven architecture
 * - Message history management
 * - Authentication support
 * - Automatic reconnection
 */

const WebSocket = require('ws');
const EventEmitter = require('events');

class CustomPumpChatClient extends EventEmitter {
  constructor(options) {
    super();
    
    // Configuration
    this.roomId = options.roomId;
    this.username = options.username || 'anonymous';
    this.messageHistoryLimit = options.messageHistoryLimit || 100;
    
    // Connection state
    this.ws = null;
    this.isConnected = false;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
    
    // Socket.IO protocol state
    this.ackId = 0;
    this.pendingAcks = new Map();
    this.pingInterval = null;
    
    // Message storage
    this.messageHistory = [];
    
    // Setup periodic cleanup
    this.cleanupInterval = setInterval(() => {
      this.cleanupStaleAcks();
    }, 10000);
  }

  /**
   * Connect to pump.fun chat server
   */
  connect() {
    const url = 'wss://livechat.pump.fun/socket.io/?EIO=4&transport=websocket';
    
    // Required headers for pump.fun
    const headers = {
      'Host': 'livechat.pump.fun',
      'Connection': 'Upgrade',
      'Pragma': 'no-cache',
      'Cache-Control': 'no-cache',
      'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36',
      'Upgrade': 'websocket',
      'Origin': 'https://pump.fun',
      'Sec-WebSocket-Version': '13',
      'Accept-Encoding': 'gzip, deflate, br, zstd',
      'Accept-Language': 'en-US,en;q=0.9',
      'Sec-WebSocket-Extensions': 'permessage-deflate; client_max_window_bits'
    };

    this.ws = new WebSocket(url, { headers });
    this.setupWebSocketHandlers();
  }

  /**
   * Setup WebSocket event handlers
   */
  setupWebSocketHandlers() {
    this.ws.on('open', () => {
      console.log('✅ WebSocket connected to pump.fun');
      this.isConnected = true;
      this.reconnectAttempts = 0;
      this.emit('connected');
    });

    this.ws.on('message', (data) => {
      const message = data.toString();
      this.handleMessage(message);
    });

    this.ws.on('close', () => {
      console.log('⚠️ WebSocket connection closed');
      this.isConnected = false;
      this.stopPing();
      this.emit('disconnected');
      this.attemptReconnect();
    });

    this.ws.on('error', (error) => {
      console.error('❌ WebSocket error:', error);
      this.emit('error', error);
    });
  }

  /**
   * Handle incoming messages based on Socket.IO protocol
   */
  handleMessage(data) {
    const messageType = data.match(/^(\d+)/)?.[1];
    
    switch (messageType) {
      case '0': // Connect message
        this.handleConnect(data);
        break;
      case '2': // Ping from server
        this.sendPong();
        break;
      case '3': // Pong from server
        // Connection is alive
        break;
      case '40': // Handshake acknowledgment
        this.handleConnectedAck(data);
        break;
      case '42': // Event message
        this.handleEvent(data);
        break;
      case '43': // Event with acknowledgment
        this.handleEventWithAck(data);
        break;
      default:
        // Handle numbered acknowledgments (430-439)
        if (messageType && messageType.startsWith('43') && messageType.length === 3) {
          this.handleNumberedAck(data);
        } else {
          console.log(`Unknown message type: ${messageType}`);
        }
    }
  }

  /**
   * Handle initial connection message
   */
  handleConnect(data) {
    const jsonData = data.substring(1);
    const connectData = JSON.parse(jsonData);
    
    if (connectData.pingInterval) {
      this.startPing(connectData.pingInterval);
    }
    
    // Send handshake
    this.send(`40{"origin":"https://pump.fun","timestamp":${Date.now()},"token":null}`);
  }

  /**
   * Handle handshake acknowledgment
   */
  handleConnectedAck(data) {
    const joinAckId = this.getNextAckId();
    this.pendingAcks.set(joinAckId, { event: 'joinRoom', timestamp: Date.now() });
    
    // Join the room
    this.send(`42${joinAckId}["joinRoom",{"roomId":"${this.roomId}","username":"${this.username}"}]`);
  }

  /**
   * Handle regular events
   */
  handleEvent(data) {
    try {
      const eventData = JSON.parse(data.substring(2));
      const [eventName, payload] = eventData;
      
      switch (eventName) {
        case 'setCookie':
          this.requestMessageHistory();
          break;
        case 'newMessage':
          this.handleNewMessage(payload);
          break;
        case 'userLeft':
          this.emit('userLeft', payload);
          break;
        default:
          console.log(`Unknown event: ${eventName}`);
      }
    } catch (error) {
      console.error('Error parsing event:', error);
    }
  }

  /**
   * Handle events with acknowledgment
   */
  handleEventWithAck(data) {
    try {
      const ackData = JSON.parse(data.substring(2));
      const eventData = ackData[0];
      
      if (eventData && eventData.messages) {
        this.messageHistory = eventData.messages;
        this.emit('messageHistory', this.messageHistory);
      } else if (Array.isArray(eventData)) {
        this.messageHistory = eventData;
        this.emit('messageHistory', this.messageHistory);
      }
    } catch (error) {
      console.error('Error parsing acknowledgment:', error);
    }
  }

  /**
   * Handle numbered acknowledgments
   */
  handleNumberedAck(data) {
    try {
      const messageType = data.match(/^(\d+)/)?.[1];
      if (!messageType) return;
      
      const ackId = parseInt(messageType.substring(2));
      const pendingAck = this.pendingAcks.get(ackId);
      
      if (pendingAck) {
        this.pendingAcks.delete(ackId);
        console.log(`✅ Received ack ${messageType} for ${pendingAck.event}`);
      }
      
      const ackData = JSON.parse(data.substring(3));
      
      if (pendingAck?.event === 'joinRoom') {
        this.requestMessageHistory();
      } else if (pendingAck?.event === 'getMessageHistory') {
        const messages = ackData[0];
        if (Array.isArray(messages)) {
          this.messageHistory = messages;
          this.emit('messageHistory', this.messageHistory);
        }
      } else if (pendingAck?.event === 'sendMessage') {
        if (ackData[0] && ackData[0].error) {
          console.error('Server error:', ackData[0]);
          this.emit('serverError', ackData[0]);
        }
      }
    } catch (error) {
      console.error('Error parsing numbered acknowledgment:', error);
    }
  }

  /**
   * Handle new chat message
   */
  handleNewMessage(message) {
    this.messageHistory.push(message);
    
    // Maintain message limit
    if (this.messageHistory.length > this.messageHistoryLimit) {
      this.messageHistory.shift();
    }
    
    this.emit('message', message);
  }

  /**
   * Request message history
   */
  requestMessageHistory() {
    const historyAckId = this.getNextAckId();
    this.pendingAcks.set(historyAckId, { event: 'getMessageHistory', timestamp: Date.now() });
    
    this.send(`42${historyAckId}["getMessageHistory",{"roomId":"${this.roomId}","before":null,"limit":${this.messageHistoryLimit}}]`);
  }

  /**
   * Send a message to the chat
   */
  sendMessage(message) {
    if (!this.isConnected) {
      console.error('Cannot send message: not connected');
      return;
    }
    
    const sendAckId = this.getNextAckId();
    this.pendingAcks.set(sendAckId, { event: 'sendMessage', timestamp: Date.now() });
    
    this.send(`42${sendAckId}["sendMessage",{"roomId":"${this.roomId}","message":"${message}","username":"${this.username}"}]`);
  }

  /**
   * Send raw data through WebSocket
   */
  send(data) {
    if (this.ws && this.isConnected) {
      this.ws.send(data);
    } else {
      console.error('Cannot send data: not connected');
    }
  }

  /**
   * Start ping interval
   */
  startPing(interval) {
    this.stopPing();
    this.pingInterval = setInterval(() => {
      this.send('2');
    }, interval);
  }

  /**
   * Stop ping interval
   */
  stopPing() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }

  /**
   * Send pong response
   */
  sendPong() {
    this.send('3');
  }

  /**
   * Get next acknowledgment ID
   */
  getNextAckId() {
    const currentId = this.ackId;
    this.ackId = (this.ackId + 1) % 10;
    return currentId;
  }

  /**
   * Cleanup stale acknowledgments
   */
  cleanupStaleAcks() {
    const now = Date.now();
    const timeout = 30000; // 30 seconds
    
    for (const [id, ack] of this.pendingAcks.entries()) {
      if (now - ack.timestamp > timeout) {
        this.pendingAcks.delete(id);
        console.log(`🧹 Cleaned up stale ack ${id} for ${ack.event}`);
      }
    }
  }

  /**
   * Attempt reconnection
   */
  attemptReconnect() {
    if (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++;
      const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
      
      console.log(`🔄 Attempting to reconnect in ${delay}ms... (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})`);
      
      setTimeout(() => {
        this.connect();
      }, delay);
    } else {
      console.error('❌ Max reconnection attempts reached');
      this.emit('maxReconnectAttemptsReached');
    }
  }

  /**
   * Disconnect from chat
   */
  disconnect() {
    this.stopPing();
    if (this.cleanupInterval) {
      clearInterval(this.cleanupInterval);
    }
    if (this.ws) {
      this.ws.close();
    }
  }

  /**
   * Get stored messages
   */
  getMessages(limit) {
    if (limit) {
      return this.messageHistory.slice(-limit);
    }
    return [...this.messageHistory];
  }

  /**
   * Get latest message
   */
  getLatestMessage() {
    return this.messageHistory[this.messageHistory.length - 1] || null;
  }

  /**
   * Check if connected
   */
  isActive() {
    return this.isConnected;
  }
}

module.exports = CustomPumpChatClient;