speed-test/src/core/botManager.js

const TelegramBot = require('node-telegram-bot-api');
const logger = require('../utils/logger');
const { Subscription } = require('../models');
const MultiSubscriptionManager = require('./multiSubscriptionManager');

class BotManager {
  constructor() {
    this.botToken = process.env.TELEGRAM_BOT_TOKEN;
    this.allowedChatIds = process.env.TELEGRAM_CHAT_ID?.split(',').map(id => id.trim()) || [];
    this.subscriptionManager = new MultiSubscriptionManager();
    this.isReconnecting = false;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
    
    // 可选的代理配置
    this.proxyConfig = null;
    if (process.env.TELEGRAM_PROXY_URL) {
      this.proxyConfig = {
        host: process.env.TELEGRAM_PROXY_HOST,
        port: parseInt(process.env.TELEGRAM_PROXY_PORT) || 1080,
        protocol: process.env.TELEGRAM_PROXY_PROTOCOL || 'http'
      };
    }
    
    if (!this.botToken) {
      logger.warn('Telegram Bot Token未配置，机器人功能将被禁用');
      this.enabled = false;
      return;
    }

    this.initializeBot();
    this.speedTestMode = 'concurrent'; // 默认测速模式
  }

  /**
   * 初始化Telegram Bot
   */
  initializeBot() {
    try {
      const botOptions = {
        polling: true,
        request: {
          timeout: 30000, // 增加超时时间到30秒
          connect_timeout: 30000
        }
      };
      
      // 如果配置了代理，添加代理设置
      if (this.proxyConfig) {
        const { HttpsProxyAgent } = require('https-proxy-agent');
        const proxyUrl = `${this.proxyConfig.protocol}://${this.proxyConfig.host}:${this.proxyConfig.port}`;
        botOptions.request.httpsAgent = new HttpsProxyAgent(proxyUrl);
        logger.info(`使用代理连接Telegram: ${proxyUrl}`);
      } else {
        botOptions.request.agent = false; // 禁用代理，避免网络问题
      }
      
      this.bot = new TelegramBot(this.botToken, botOptions);
      
      // 改进的错误处理
      this.bot.on('polling_error', (error) => {
        logger.warn('Telegram轮询错误，尝试重连...', { 
          error: error.message,
          code: error.code 
        });
        
        // 处理各种网络错误
        const isNetworkError = error.code === 'EFATAL' || 
                              error.code === 'ESOCKETTIMEDOUT' ||
                              error.code === 'ECONNRESET' ||
                              error.code === 'ENOTFOUND' ||
                              error.message.includes('timeout') ||
                              error.message.includes('network') ||
                              error.message.includes('AggregateError');
        
        if (isNetworkError) {
          this.handleReconnection();
        }
      });
      
      // 添加webhook错误处理
      this.bot.on('webhook_error', (error) => {
        logger.warn('Telegram Webhook错误', { 
          error: error.message,
          code: error.code 
        });
      });
      
      // 添加错误处理
      this.bot.on('error', (error) => {
        logger.error('Telegram Bot错误', { 
          error: error.message,
          code: error.code 
        });
      });
      
      this.enabled = true;
      this.setupCommands();
      this.setupMessageHandlers();
      logger.info('Telegram机器人管理器初始化成功');
    } catch (error) {
      logger.error('Telegram机器人管理器初始化失败', { error: error.message });
      this.enabled = false;
    }
  }

  /**
   * 处理重连逻辑
   */
  handleReconnection() {
    // 防止重复重连
    if (this.isReconnecting) {
      logger.debug('Telegram重连正在进行中，跳过本次重连');
      return;
    }
    
    this.isReconnecting = true;
    this.reconnectAttempts++;
    
    if (this.reconnectAttempts > this.maxReconnectAttempts) {
      logger.error('Telegram重连次数超过限制，停止重连', {
        attempts: this.reconnectAttempts,
        maxAttempts: this.maxReconnectAttempts
      });
      this.isReconnecting = false;
      this.reconnectAttempts = 0;
      return;
    }
    
    const delay = Math.min(10000 * this.reconnectAttempts, 60000); // 递增延迟，最大60秒
    
    setTimeout(async () => {
      try {
        logger.info(`尝试重新连接Telegram Bot... (第${this.reconnectAttempts}次尝试)`);
        
        // 停止当前轮询
        if (this.bot && this.bot.stopPolling) {
          this.bot.stopPolling();
        }
        
        // 等待一段时间后重新启动
        setTimeout(async () => {
          try {
            if (this.bot && this.bot.startPolling) {
              await this.bot.startPolling();
              logger.info('Telegram机器人重连成功');
              this.isReconnecting = false;
              this.reconnectAttempts = 0; // 重置重连计数
            }
          } catch (reconnectError) {
            logger.error('Telegram机器人重连失败', { 
              error: reconnectError.message,
              code: reconnectError.code,
              attempt: this.reconnectAttempts
            });
            
            // 如果重连失败，再次尝试
            this.isReconnecting = false;
            if (this.reconnectAttempts < this.maxReconnectAttempts) {
              this.handleReconnection();
            }
          }
        }, 5000);
        
      } catch (error) {
        logger.error('Telegram重连过程中发生错误', { error: error.message });
        this.isReconnecting = false;
      }
    }, delay);
  }

  /**
   * 设置机器人命令
   */
  setupCommands() {
    if (!this.enabled) return;

    this.bot.setMyCommands([
      { command: '/start', description: '开始使用机器人' },
      { command: '/help', description: '显示帮助信息' },
      { command: '/status', description: '查看系统状态' },
      { command: '/subscriptions', description: '查看所有订阅' },
      { command: '/add_subscription', description: '添加订阅链接' },
      { command: '/remove_subscription', description: '删除订阅' },
      { command: '/update_subscriptions', description: '手动更新订阅' },
      { command: '/test_speed', description: '手动触发测速' },
      { command: '/set_speed_mode', description: '设置测速模式（concurrent/serial）' }
    ]);
  }

  /**
   * 设置消息处理器
   */
  setupMessageHandlers() {
    if (!this.enabled) return;

    // 处理 /start 命令
    this.bot.onText(/\/start/, async (msg) => {
      await this.handleStart(msg);
    });

    // 处理 /help 命令
    this.bot.onText(/\/help/, async (msg) => {
      await this.handleHelp(msg);
    });

    // 处理 /status 命令
    this.bot.onText(/\/status/, async (msg) => {
      await this.handleStatus(msg);
    });

    // 处理 /subscriptions 命令
    this.bot.onText(/\/subscriptions/, async (msg) => {
      await this.handleSubscriptions(msg);
    });

    // 处理 /add_subscription 命令
    this.bot.onText(/\/add_subscription/, async (msg) => {
      await this.handleAddSubscription(msg);
    });

    // 处理 /remove_subscription 命令
    this.bot.onText(/\/remove_subscription/, async (msg) => {
      await this.handleRemoveSubscription(msg);
    });

    // 处理删除订阅的具体命令
    this.bot.onText(/\/delete_(\d+)/, async (msg, match) => {
      await this.handleDeleteSubscription(msg, parseInt(match[1]));
    });

    // 处理 /update_subscriptions 命令
    this.bot.onText(/\/update_subscriptions/, async (msg) => {
      await this.handleUpdateSubscriptions(msg);
    });

    // 处理 /test_speed 命令
    this.bot.onText(/\/test_speed/, async (msg) => {
      await this.handleTestSpeed(msg);
    });

    // 处理 /set_speed_mode 命令
    this.bot.onText(/\/set_speed_mode (.+)/, async (msg, match) => {
      await this.handleSetSpeedMode(msg, match[1]);
    });

    // 处理普通消息（只处理订阅链接，不处理其他消息）
    this.bot.on('message', async (msg) => {
      // 只处理文本消息
      if (!msg.text) return;
      
      // 如果是命令，不处理（由命令处理器处理）
      if (msg.text.startsWith('/')) return;
      
      // 只处理订阅链接
      if (this.isSubscriptionUrl(msg.text)) {
        await this.addSubscription(msg.chat.id, msg.text);
      }
      // 其他消息不回复
    });
  }

  /**
   * 检查用户权限
   */
  isAuthorized(chatId) {
    return this.allowedChatIds.includes(chatId.toString());
  }

  /**
   * 发送消息
   */
  async sendMessage(chatId, message, options = {}) {
    try {
      return await this.bot.sendMessage(chatId, message, {
        parse_mode: 'Markdown',
        disable_web_page_preview: true,
        ...options
      });
    } catch (error) {
      logger.error('发送Telegram消息失败', { error: error.message, chatId });
      throw error;
    }
  }

  /**
   * 处理 /start 命令
   */
  async handleStart(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    const message = `🤖 *欢迎使用测速机器人！*

📋 *可用命令：*
• /help - 显示帮助信息
• /status - 查看系统状态
• /subscriptions - 查看所有订阅
• /add_subscription - 添加订阅链接
• /remove_subscription - 删除订阅
• /update_subscriptions - 手动更新订阅
• /test_speed - 手动触发测速

💡 *使用提示：*
• 直接发送订阅链接即可添加订阅
• 使用 /remove_subscription 查看订阅列表
• 使用 /delete_1 删除第1个订阅
• 支持多种格式：Clash、Shadowsocks、Vmess等`;

    await this.sendMessage(chatId, message);
  }

  /**
   * 处理 /help 命令
   */
  async handleHelp(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    const message = `📖 *帮助信息*

🔧 *订阅管理：*
• 直接发送订阅链接即可添加
• 支持格式：Clash、Shadowsocks、Vmess、Trojan
• 系统会自动解析并去重

⚡ *测速功能：*
• 自动定时测速（默认每10分钟）
• 支持手动触发测速
• 节点故障自动通知

📊 *状态监控：*
• 实时监控节点状态
• 延迟和速度测试
• 故障节点自动标记

🔔 *通知功能：*
• 节点故障通知
• 节点恢复通知
• 测试摘要报告

💡 *使用示例：*
• 发送：\`https://example.com/subscription\`
• 发送：\`ss://...\`
• 发送：\`vmess://...\`
• 删除：\`/delete_1\` 删除第1个订阅`;

    await this.sendMessage(chatId, message);
  }

  /**
   * 处理 /status 命令
   */
  async handleStatus(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    try {
      const status = await this.subscriptionManager.getStatus();
      const subscriptions = await Subscription.findAll({
        where: { isActive: true }
      });

      const totalNodes = subscriptions.reduce((sum, sub) => sum + (sub.nodeCount || 0), 0);
      
      const message = `📊 *系统状态*

📡 *订阅信息：*
• 活跃订阅：${subscriptions.length} 个
• 总节点数：${totalNodes} 个

🔄 *更新状态：*
• 自动更新：${status.autoUpdateEnabled ? '✅ 启用' : '❌ 禁用'}
• 更新间隔：${status.updateInterval || '未设置'} 秒
• 最后更新：${status.lastUpdateTime || '未更新'}

⚡ *测速状态：*
• 测速触发器：${status.speedTestTrigger ? '✅ 已设置' : '❌ 未设置'}
• 定时测速：${process.env.ENABLE_SCHEDULED_SPEED_TEST !== 'false' ? '✅ 启用' : '❌ 禁用'}

🛠️ *系统信息：*
• 运行时间：${this.getUptime()}
• 内存使用：${this.getMemoryUsage()}`;

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('获取状态失败', { error: error.message });
      await this.sendMessage(chatId, '❌ 获取系统状态失败');
    }
  }

  /**
   * 处理 /subscriptions 命令
   */
  async handleSubscriptions(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    try {
      const subscriptions = await Subscription.findAll({
        where: { isActive: true },
        order: [['createdAt', 'ASC']]
      });

      if (subscriptions.length === 0) {
        await this.sendMessage(chatId, '📭 暂无订阅，请使用 /add_subscription 添加订阅');
        return;
      }

      let message = `📡 *订阅列表*\n\n`;
      
      subscriptions.forEach((sub, index) => {
        const status = sub.isActive ? '✅' : '❌';
        const lastUpdate = sub.lastUpdateTime 
          ? new Date(sub.lastUpdateTime).toLocaleString('zh-CN')
          : '未更新';
        
        message += `${index + 1}. ${status} *${sub.name}*\n`;
        message += `   📊 节点数：${sub.nodeCount || 0}\n`;
        message += `   🔗 URL：\`${sub.url}\`\n`;
        message += `   ⏰ 最后更新：${lastUpdate}\n\n`;
      });

      message += `💡 *操作提示：*\n`;
      message += `• 使用 /remove_subscription 删除订阅\n`;
      message += `• 使用 /update_subscriptions 手动更新\n`;
      message += `• 直接发送新链接添加订阅`;

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('获取订阅列表失败', { error: error.message });
      await this.sendMessage(chatId, '❌ 获取订阅列表失败');
    }
  }

  /**
   * 处理 /add_subscription 命令
   */
  async handleAddSubscription(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    await this.sendMessage(chatId, 
      `📥 *添加订阅*\n\n` +
      `请发送订阅链接，支持以下格式：\n` +
      `• Clash配置：\`https://example.com/clash.yaml\`\n` +
      `• Shadowsocks：\`ss://...\`\n` +
      `• Vmess：\`vmess://...\`\n` +
      `• Trojan：\`trojan://...\`\n\n` +
      `💡 直接发送链接即可添加订阅`
    );
  }

  /**
   * 处理 /remove_subscription 命令
   */
  async handleRemoveSubscription(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    try {
      const subscriptions = await Subscription.findAll({
        where: { isActive: true },
        order: [['createdAt', 'ASC']]
      });

      if (subscriptions.length === 0) {
        await this.sendMessage(chatId, '📭 暂无订阅可删除');
        return;
      }

      let message = `🗑️ *删除订阅*\n\n`;
      message += `请选择要删除的订阅：\n\n`;
      
      subscriptions.forEach((sub, index) => {
        message += `${index + 1}. *${sub.name}*\n`;
        message += `   📊 节点数：${sub.nodeCount || 0}\n`;
        message += `   🔗 \`${sub.url}\`\n\n`;
      });

      message += `💡 使用 /delete_1 删除第1个订阅\n`;
      message += `💡 使用 /delete_2 删除第2个订阅\n`;
      message += `💡 以此类推...`;

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('获取订阅列表失败', { error: error.message });
      await this.sendMessage(chatId, '❌ 获取订阅列表失败');
    }
  }

  /**
   * 处理 /update_subscriptions 命令
   */
  async handleUpdateSubscriptions(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    try {
      await this.sendMessage(chatId, '🔄 开始手动更新订阅...');
      
      const results = await this.subscriptionManager.manualUpdate();
      
      let message = `✅ *订阅更新完成*\n\n`;
      
      results.forEach(result => {
        if (result.error) {
          message += `❌ *${result.subscriptionName}*\n`;
          message += `   错误：${result.error}\n\n`;
        } else {
          message += `✅ *${result.subscriptionName}*\n`;
          message += `   新增：${result.added} 个\n`;
          message += `   更新：${result.updated} 个\n`;
          message += `   移除：${result.removed} 个\n\n`;
        }
      });

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('手动更新订阅失败', { error: error.message });
      await this.sendMessage(chatId, '❌ 更新订阅失败：' + error.message);
    }
  }

  /**
   * 处理 /test_speed 命令
   */
  async handleTestSpeed(msg) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    try {
      await this.sendMessage(chatId, `⚡ 开始手动测速...\n当前测速模式: *${this.speedTestMode === 'concurrent' ? '并发连通性' : '串行精准延迟'}*`);
      // 获取所有节点（示例，实际请根据你的业务逻辑获取节点列表）
      const subscriptions = await Subscription.findAll({ where: { isActive: true } });
      let allNodes = [];
      for (const sub of subscriptions) {
        if (sub.nodes) {
          allNodes = allNodes.concat(sub.nodes);
        }
      }
      if (allNodes.length === 0) {
        await this.sendMessage(chatId, '❌ 当前没有可用节点');
        return;
      }
      let results;
      if (this.speedTestMode === 'concurrent') {
        // 并发连通性测试
        const SpeedTester = require('./speedTester');
        const speedTester = new SpeedTester();
        results = await speedTester.testNodes(allNodes, { concurrency: 20 });
      } else {
        // 串行精准延迟测试
        const SpeedTester = require('./speedTester');
        const speedTester = new SpeedTester();
        results = await speedTester.testNodesSerial(allNodes);
      }
      // 简要汇总结果
      const success = results.filter(r => r.isSuccess);
      const failed = results.filter(r => !r.isSuccess);
      let msgText = `测速完成！\n成功: ${success.length} 个, 失败: ${failed.length} 个`;
      if (success.length > 0) {
        msgText += `\n前5个成功节点: ` + success.slice(0, 5).map(r => `${r.nodeName || r.nodeId}(${r.latency}ms)`).join(', ');
      }
      if (failed.length > 0) {
        msgText += `\n前5个失败节点: ` + failed.slice(0, 5).map(r => r.nodeName || r.nodeId).join(', ');
      }
      await this.sendMessage(chatId, msgText);
    } catch (error) {
      logger.error('手动测速失败', { error: error.message });
      await this.sendMessage(chatId, '❌ 测速失败：' + error.message);
    }
  }

  /**
   * 处理测速模式设置命令
   */
  async handleSetSpeedMode(msg, mode) {
    const chatId = msg.chat.id;
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }
    const validModes = ['concurrent', 'serial'];
    if (!validModes.includes(mode)) {
      await this.sendMessage(chatId, '❌ 无效的测速模式。可选: concurrent（并发连通性）, serial（串行精准延迟）');
      return;
    }
    this.speedTestMode = mode;
    await this.sendMessage(chatId, `✅ 已切换测速模式为: *${mode === 'concurrent' ? '并发连通性' : '串行精准延迟'}*`);
  }

  /**
   * 检查是否是订阅链接
   */
  isSubscriptionUrl(text) {
    return /^(https?:\/\/|ss:\/\/|vmess:\/\/|trojan:\/\/)/.test(text);
  }

  /**
   * 添加订阅
   */
  async addSubscription(chatId, url) {
    try {
      await this.sendMessage(chatId, '📥 正在添加订阅...');
      
      // 创建新订阅
      const subscription = await Subscription.create({
        name: `订阅_${Date.now()}`,
        url: url,
        isActive: true,
        nodeCount: 0
      });

      // 更新订阅
      const result = await this.subscriptionManager.manualUpdateSubscription(subscription.id);
      
      if (result.error) {
        await this.sendMessage(chatId, `❌ 添加订阅失败：${result.error}`);
        return;
      }

      const message = `✅ *订阅添加成功*\n\n` +
        `📡 订阅名称：*${subscription.name}*\n` +
        `🔗 URL：\`${url}\`\n` +
        `📊 节点数：${result.actualNodeCount || 0}\n` +
        `➕ 新增：${result.added || 0} 个\n` +
        `🔄 更新：${result.updated || 0} 个\n` +
        `🗑️ 移除：${result.removed || 0} 个`;

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('添加订阅失败', { error: error.message, url });
      await this.sendMessage(chatId, '❌ 添加订阅失败：' + error.message);
    }
  }

  /**
   * 处理删除订阅命令
   */
  async handleDeleteSubscription(msg, index) {
    const chatId = msg.chat.id;
    
    if (!this.isAuthorized(chatId)) {
      await this.sendMessage(chatId, '❌ 您没有权限使用此机器人');
      return;
    }

    try {
      const subscriptions = await Subscription.findAll({
        where: { isActive: true },
        order: [['createdAt', 'DESC']]
      });

      if (index < 1 || index > subscriptions.length) {
        await this.sendMessage(chatId, '❌ 无效的订阅编号');
        return;
      }

      const subscription = subscriptions[index - 1];
      
      // 标记为非活跃
      await subscription.update({ isActive: false });
      
      const message = `🗑️ *订阅删除成功*\n\n` +
        `📡 订阅名称：*${subscription.name}*\n` +
        `🔗 URL：\`${subscription.url}\`\n` +
        `📊 节点数：${subscription.nodeCount || 0}`;

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('删除订阅失败', { error: error.message, index });
      await this.sendMessage(chatId, '❌ 删除订阅失败：' + error.message);
    }
  }

  /**
   * 根据索引删除订阅（保留兼容性）
   */
  async removeSubscriptionByIndex(chatId, index) {
    try {
      const subscriptions = await Subscription.findAll({
        where: { isActive: true },
        order: [['createdAt', 'DESC']]
      });

      if (index < 1 || index > subscriptions.length) {
        await this.sendMessage(chatId, '❌ 无效的订阅编号');
        return;
      }

      const subscription = subscriptions[index - 1];
      
      // 标记为非活跃
      await subscription.update({ isActive: false });
      
      const message = `🗑️ *订阅删除成功*\n\n` +
        `📡 订阅名称：*${subscription.name}*\n` +
        `🔗 URL：\`${subscription.url}\`\n` +
        `📊 节点数：${subscription.nodeCount || 0}`;

      await this.sendMessage(chatId, message);
    } catch (error) {
      logger.error('删除订阅失败', { error: error.message, index });
      await this.sendMessage(chatId, '❌ 删除订阅失败：' + error.message);
    }
  }

  /**
   * 获取运行时间
   */
  getUptime() {
    const uptime = process.uptime();
    const hours = Math.floor(uptime / 3600);
    const minutes = Math.floor((uptime % 3600) / 60);
    return `${hours}小时${minutes}分钟`;
  }

  /**
   * 获取内存使用情况
   */
  getMemoryUsage() {
    const usage = process.memoryUsage();
    const used = Math.round(usage.heapUsed / 1024 / 1024);
    const total = Math.round(usage.heapTotal / 1024 / 1024);
    return `${used}MB / ${total}MB`;
  }

  /**
   * 停止机器人
   */
  stop() {
    if (this.enabled && this.bot) {
      this.bot.stopPolling();
      logger.info('Telegram机器人已停止');
    }
  }
}

module.exports = BotManager;