PUMPFUN TTS

pumpfun-chat-tts.js

/**
 * Pumpfun Chat → ElevenLabs TTS Bot with Queue
 *
 * Features:
 *   - Automatically converts ANY message in chat to speech
 *   - Uses default voice for all messages
 *   - Skips bot's own messages to avoid loops
 *   - Filters out very short messages and non-text content
 *   - Queues multiple TTS requests for smooth playback
 *
 * Requirements:
 *   bun install ws events @elevenlabs/elevenlabs-js dotenv speaker
 */

import WebSocket from 'ws';
import EventEmitter from 'events';
import { ElevenLabsClient } from '@elevenlabs/elevenlabs-js';
import dotenv from 'dotenv';
import playSound from 'play-sound';
// @ts-ignore - node-wav-player doesn't have types
import wavPlayer from 'node-wav-player';

dotenv.config();

// Types
interface TTSQueueItem {
  voice: string;
  text: string;
  user: string;
}

interface Voice {
  voice_id: string;
  name: string;
  category: string;
}


// --- Pumpfun Chat Client (Raw WebSocket with Fixed Protocol) ---
class PumpChatClient extends EventEmitter {
  private roomId: string;
  private username: string;
  private messageHistoryLimit: number;
  private ws: WebSocket | null = null;
  private isConnected: boolean = false;
  private reconnectAttempts: number = 0;
  private maxReconnectAttempts: number = 5;
  private ackId: number = 0;
  private pendingAcks: Map<number, { event: string; timestamp: number }> = new Map();
  private pingInterval: NodeJS.Timeout | null = null;
  private messageHistory: any[] = [];
  private cleanupInterval: NodeJS.Timeout;

  constructor({ roomId, username, messageHistoryLimit = 100 }: {
    roomId: string;
    username: string;
    messageHistoryLimit?: number;
  }) {
    super();
    this.roomId = roomId;
    this.username = username;
    this.messageHistoryLimit = messageHistoryLimit;
    
    // Setup periodic cleanup
    this.cleanupInterval = setInterval(() => {
      this.cleanupStaleAcks();
    }, 10000);
  }

  connect() {
    const url = 'wss://livechat.pump.fun/socket.io/?EIO=4&transport=websocket';
    
    // Minimal headers for pump.fun
    const headers = {
      'Origin': 'https://pump.fun',
      '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'
    };

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

  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();
      console.log('📨 Received message:', message);
      this.handleMessage(message);
    });

    this.ws!.on('close', (code, reason) => {
      console.log(`⚠️ WebSocket connection closed - Code: ${code}, Reason: ${reason}`);
      this.isConnected = false;
      this.stopPing();
      this.emit('disconnected');
      this.attemptReconnect();
    });

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

  handleMessage(data: string) {
    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}`);
        }
    }
  }

  handleConnect(data: string) {
    const jsonData = data.substring(1);
    const connectData = JSON.parse(jsonData);
    
    console.log('🔗 Socket.IO handshake received:', connectData);
    
    if (connectData.pingInterval) {
      this.startPing(connectData.pingInterval);
    }
    
    // Send handshake acknowledgment
    console.log('📤 Sending handshake acknowledgment...');
    this.send(`40{"origin":"https://pump.fun","timestamp":${Date.now()},"token":null}`);
  }

  handleConnectedAck(data: string) {
    console.log('✅ Handshake acknowledged, joining room...');
    const joinAckId = this.getNextAckId();
    this.pendingAcks.set(joinAckId, { event: 'joinRoom', timestamp: Date.now() });
    
    // Join the room
    console.log(`📤 Joining room: ${this.roomId} as ${this.username}`);
    this.send(`42${joinAckId}["joinRoom",{"roomId":"${this.roomId}","username":"${this.username}"}]`);
  }

  handleEvent(data: string) {
    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);
    }
  }

  handleEventWithAck(data: string) {
    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);
    }
  }

  handleNumberedAck(data: string) {
    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);
    }
  }

  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}}]`);
  }

  handleNewMessage(message: any) {
    this.messageHistory.push(message);
    
    // Maintain message limit
    if (this.messageHistory.length > this.messageHistoryLimit) {
      this.messageHistory.shift();
    }
    
    this.emit('message', message);
  }

  sendMessage(message: string) {
    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(data: string) {
    if (this.ws && this.isConnected) {
      console.log('📤 Sending:', data);
      this.ws.send(data);
    } else {
      console.error('❌ Cannot send data: not connected');
    }
  }

  startPing(interval: number) {
    this.stopPing();
    this.pingInterval = setInterval(() => {
      this.send('2');
    }, interval);
  }

  stopPing() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }

  sendPong() {
    this.send('3');
  }

  getNextAckId(): number {
    const currentId = this.ackId;
    this.ackId = (this.ackId + 1) % 10;
    return currentId;
  }

  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}`);
      }
    }
  }

  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() {
    this.stopPing();
    if (this.cleanupInterval) {
      clearInterval(this.cleanupInterval);
    }
    if (this.ws) {
      this.ws.close();
    }
  }

  /**
   * Get stored messages
   */
  getMessages(limit?: number) {
    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;
  }
}

// --- ElevenLabs ---
const ELEVENLABS_API_KEY = process.env.ELEVENLABS_API_KEY;
if (!ELEVENLABS_API_KEY) throw new Error('❌ Missing ELEVENLABS_API_KEY in .env');

const elevenlabs = new ElevenLabsClient({ apiKey: ELEVENLABS_API_KEY });

let voicesCache: Voice[] = [];
const MODEL_ID = 'eleven_multilingual_v2';

// --- Audio Playback ---
const player = playSound({});

async function playAudio(audioData: Buffer, format: string = 'pcm_16000'): Promise<void> {
  return new Promise(async (resolve, reject) => {
    // Create a temporary file for audio playback
    const tempFile = `/tmp/tts_${Date.now()}.wav`;
    
    try {
      // Convert PCM data to WAV format
      const wavHeader = createWavHeader(audioData.length, 16000, 1, 16);
      const wavData = Buffer.concat([wavHeader, audioData]);
      
      // Write to temporary file
      Bun.write(tempFile, wavData);
      
      // Try different playback methods
      const playMethods = [
        // Method 1: play-sound with afplay (macOS)
        () => {
          return new Promise<void>((resolve, reject) => {
            player.play(tempFile, { afplay: ['-v', '0.5'] }, (err: any) => {
              if (err) reject(err);
              else resolve();
            });
          });
        },
        
        // Method 2: node-wav-player
        () => wavPlayer.play({ path: tempFile }),
        
        // Method 3: play-sound default
        () => {
          return new Promise<void>((resolve, reject) => {
            player.play(tempFile, (err: any) => {
              if (err) reject(err);
              else resolve();
            });
          });
        }
      ];
      
      // Try each method until one works
      let lastError: any = null;
      for (const method of playMethods) {
        try {
          await method();
          // Clean up temp file
          Bun.file(tempFile).unlink().catch(() => {});
          resolve();
          return;
        } catch (err) {
          lastError = err;
          continue;
        }
      }
      
      // If all methods fail, clean up and reject
      Bun.file(tempFile).unlink().catch(() => {});
      reject(lastError);
      
    } catch (err) {
      // Clean up temp file on error
      Bun.file(tempFile).unlink().catch(() => {});
      reject(err);
    }
  });
}

function createWavHeader(dataLength: number, sampleRate: number, channels: number, bitsPerSample: number): Buffer {
  const header = Buffer.alloc(44);
  
  // RIFF header
  header.write('RIFF', 0);
  header.writeUInt32LE(36 + dataLength, 4);
  header.write('WAVE', 8);
  
  // fmt chunk
  header.write('fmt ', 12);
  header.writeUInt32LE(16, 16); // fmt chunk size
  header.writeUInt16LE(1, 20);  // audio format (PCM)
  header.writeUInt16LE(channels, 22);
  header.writeUInt32LE(sampleRate, 24);
  header.writeUInt32LE(sampleRate * channels * bitsPerSample / 8, 28); // byte rate
  header.writeUInt16LE(channels * bitsPerSample / 8, 32); // block align
  header.writeUInt16LE(bitsPerSample, 34);
  
  // data chunk
  header.write('data', 36);
  header.writeUInt32LE(dataLength, 40);
  
  return header;
}

// --- TTS Queue ---
const ttsQueue: TTSQueueItem[] = [];
let isPlaying = false;

async function enqueueTTS(voiceId: string, text: string, user: string) {
  ttsQueue.push({ voice: voiceId, text, user });
  if (!isPlaying) {
    processQueue();
  }
}

async function processQueue() {
  if (ttsQueue.length === 0) {
    isPlaying = false;
    return;
  }

  isPlaying = true;
  const { voice, text, user } = ttsQueue.shift()!;
  console.log(`🔊 Playing TTS (${voice}) from ${user}: "${text}"`);

  try {
    const audioStream = await elevenlabs.textToSpeech.stream(voice, {
      modelId: MODEL_ID,
      text,
      outputFormat: 'pcm_16000',
      voiceSettings: {
        stability: 0.3,
        similarityBoost: 0.8,
        useSpeakerBoost: true,
        speed: 1.0,
      },
    });

    // Collect audio data and play it
    let audioData = Buffer.alloc(0);
    for await (const chunk of audioStream) {
      audioData = Buffer.concat([audioData, chunk]);
    }
    
    console.log(`🎵 Generated ${audioData.length} bytes of audio data`);
    
    // Play the audio through speakers
    try {
      await playAudio(audioData);
      console.log(`🔊 Successfully played TTS audio`);
    } catch (audioError) {
      console.error('❌ Audio playback error:', audioError);
      // Continue execution even if audio fails
    }

    // TTS completed successfully - no need to spam chat
    console.log(`✅ TTS completed for ${user}: "${text}"`);
  } catch (err) {
    console.error('❌ TTS error:', err);
    // Error logged to console - no message sent to chat
  }

  // Play next after finishing
  setImmediate(processQueue);
}

// --- Configuration ---
// Allow command line argument for token address: bun run bot.ts <token_address>
const args = process.argv.slice(2);
const TOKEN_ADDRESS = args[0] || process.env.TOKEN_ADDRESS || 'YOUR_TOKEN_ADDRESS';
const BOT_USERNAME = process.env.BOT_USERNAME || 'tts-bot';

// Validate token address format (Solana addresses are 32-44 characters, base58)
function validateTokenAddress(address: string): boolean {
  if (address === 'YOUR_TOKEN_ADDRESS') {
    console.error('❌ Please set a valid TOKEN_ADDRESS in your .env file');
    return false;
  }
  
  // Basic Solana address validation (base58, 32-44 characters)
  const base58Regex = /^[1-9A-HJ-NP-Za-km-z]{32,44}$/;
  if (!base58Regex.test(address)) {
    console.error('❌ Invalid token address format. Solana addresses should be 32-44 base58 characters');
    return false;
  }
  
  return true;
}

if (!validateTokenAddress(TOKEN_ADDRESS)) {
  console.error('❌ Invalid token address. Please set TOKEN_ADDRESS in your .env file');
  console.error('   Example: TOKEN_ADDRESS=So11111111111111111111111111111111111111112');
  process.exit(1);
}

// --- Pumpfun bot setup ---
const pumpClient = new PumpChatClient({
  roomId: TOKEN_ADDRESS,
  username: BOT_USERNAME,
});

pumpClient.on('connected', async () => {
  console.log(`✅ Connected to Pumpfun chat for token: ${TOKEN_ADDRESS}`);
  console.log(`🤖 Bot username: ${BOT_USERNAME}`);
  console.log(`📊 Message logging enabled - all chat messages will be displayed`);
  console.log(`🔇 Listen-only mode: Bot will not send any messages to chat`);
  await loadVoices();
});

// Room joined successfully
pumpClient.on('messageHistory', async () => {
  console.log('🎉 Successfully joined room! Ready to listen for messages...');
});

pumpClient.on('message', async (msg: any) => {
  const text = msg?.message || msg?.text || '';
  const user = msg?.username || 'unknown';
  
  // Log every message received
  console.log(`📨 [${new Date().toLocaleTimeString()}] ${user}: "${text}"`);
  
  // Skip empty messages
  if (!text.trim()) {
    console.log(`⏭️  Skipping empty message from ${user}`);
    return;
  }
  
  // Skip messages from the bot itself to avoid loops
  if (user === BOT_USERNAME) {
    console.log(`⏭️  Skipping bot's own message`);
    return;
  }
  
  // Skip very short messages (like single characters or emojis)
  if (text.trim().length < 2) {
    console.log(`⏭️  Skipping short message from ${user}: "${text}"`);
    return;
  }
  
  // Skip messages that are just punctuation or numbers
  if (!/[a-zA-Z]/.test(text)) {
    console.log(`⏭️  Skipping non-text message from ${user}: "${text}"`);
    return;
  }

  console.log(`🎤 Converting message from ${user} to TTS: "${text}"`);
  
  // Load voices if not already loaded
  if (voicesCache.length === 0) {
    await loadVoices();
  }
  
  // Randomly select a voice for each message
  const chosenVoice = getRandomVoice();
  
  // Queue the TTS
  await enqueueTTS(chosenVoice, text, user);
});

// --- Helpers ---
async function loadVoices() {
  try {
    const req = await fetch('https://api.elevenlabs.io/v1/voices', {
      headers: { 'xi-api-key': ELEVENLABS_API_KEY! },
    });
    const data = await req.json() as { voices?: Voice[] };
    voicesCache = data.voices || [];
    console.log(`✅ Loaded ${voicesCache.length} voices from ElevenLabs`);
    
    // Log voice categories for debugging
    const categories = [...new Set(voicesCache.map(v => v.category))];
    console.log(`📊 Available voice categories: ${categories.join(', ')}`);
  } catch (err) {
    console.error('❌ Failed to fetch voices:', err);
    voicesCache = [];
  }
}

function getRandomVoice(): string {
  if (voicesCache.length === 0) {
    console.warn('⚠️  No voices available, using fallback voice');
    return 'JBFqnCBsd6RMkjVDRZzb'; // Fallback voice
  }
  
  const randomIndex = Math.floor(Math.random() * voicesCache.length);
  const selectedVoice = voicesCache[randomIndex];
  
  if (!selectedVoice) {
    console.warn('⚠️  Selected voice is undefined, using fallback voice');
    return 'JBFqnCBsd6RMkjVDRZzb'; // Fallback voice
  }
  
  console.log(`🎲 Randomly selected voice: ${selectedVoice.name} (${selectedVoice.voice_id})`);
  return selectedVoice.voice_id;
}

// start
console.log('🚀 Starting TTS Bot (Listen-Only Mode)...');
console.log('📋 Configuration:');
console.log(`   Token Address: ${TOKEN_ADDRESS}`);
console.log(`   Bot Username: ${BOT_USERNAME}`);
console.log(`   ElevenLabs API: ${ELEVENLABS_API_KEY ? '✅ Configured' : '❌ Missing'}`);
console.log('🔇 Mode: Listen-only (no messages will be sent to chat)');
console.log('📊 All chat messages will be logged below:');
console.log('─'.repeat(50));

pumpClient.connect();