executor.js

const https = require('https');
const http = require('http');
const log = require('./logger');

const USER_AGENTS = [
  'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
  'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
  'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
  'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.1 Safari/605.1.15',
  'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:132.0) Gecko/20100101 Firefox/132.0',
  'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:132.0) Gecko/20100101 Firefox/132.0',
  'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
  'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36 Edg/131.0.0.0'
];

const ACCEPT_LANGUAGES = [
  'en-US,en;q=0.9',
  'en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7',
  'zh-CN,zh;q=0.9,en;q=0.8',
  'en-GB,en;q=0.9,en-US;q=0.8',
  'zh-TW,zh;q=0.9,en-US;q=0.8,en;q=0.7'
];

function getRandomUserAgent() {
  return USER_AGENTS[Math.floor(Math.random() * USER_AGENTS.length)];
}

function getRandomAcceptLanguage() {
  return ACCEPT_LANGUAGES[Math.floor(Math.random() * ACCEPT_LANGUAGES.length)];
}

function buildDisguiseHeaders() {
  return {
    'User-Agent': getRandomUserAgent(),
    'Accept': 'application/json, text/event-stream, */*',
    'Accept-Language': getRandomAcceptLanguage(),
    'Accept-Encoding': 'gzip, deflate, br',
    'Cache-Control': 'no-cache',
    'Pragma': 'no-cache',
    'Sec-Ch-Ua': '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
    'Sec-Ch-Ua-Mobile': '?0',
    'Sec-Ch-Ua-Platform': '"Windows"',
    'Sec-Fetch-Dest': 'empty',
    'Sec-Fetch-Mode': 'cors',
    'Sec-Fetch-Site': 'cross-site',
    'Connection': 'keep-alive'
  };
}

function buildUrl(baseUrl, endpoint) {
  let url = baseUrl.trim().replace(/\/ $/, '');
  if (!url) throw new Error('Base URL 不能为空');
  if (!url.endsWith('/v1')) url += '/v1';
  return url + endpoint;
}

function randHex(n) {
  const chars = '0123456789abcdef';
  let result = '';
  for (let i = 0; i < n; i++) {
    result += chars[Math.floor(Math.random() * 16)];
  }
  return result;
}

function randText(n) {
  const len = Math.max(0, Math.min(4000, Math.floor(Number(n) || 0)));
  if (len <= 0) return '';
  return randHex(Math.ceil(len / 2)).slice(0, len);
}

function buildUserPrompt(text, useRand, markerLen) {
  if (!useRand) return { prompt: text, tag: null, marker: null };
  const tag = 'r' + randText(markerLen);
  const marker = '[RANDOM_MARKER:' + tag + ']';
  return { prompt: marker + text, tag, marker };
}

function estimateTokens(text) {
  return Math.max(1, Math.ceil(String(text || '').length / 4));
}

function usageWithFallback(usage, reqPrompt, sysPrompt, respText) {
  let p = Number(usage?.prompt_tokens || 0);
  let c = Number(usage?.completion_tokens || 0);
  let t = Number(usage?.total_tokens || 0);
  let estimated = false;

  if (p <= 0 && c <= 0 && t <= 0) {
    p = estimateTokens((sysPrompt ? sysPrompt + '\n' : '') + (reqPrompt || ''));
    c = estimateTokens(respText || '');
    t = p + c;
    estimated = true;
  } else {
    if (p <= 0) { p = estimateTokens((sysPrompt ? sysPrompt + '\n' : '') + (reqPrompt || '')); estimated = true; }
    if (c <= 0) { c = estimateTokens(respText || ''); estimated = true; }
    if (t <= 0) { t = p + c; estimated = true; }
  }
  return { prompt_tokens: p, completion_tokens: c, total_tokens: t, estimated };
}

function makeRequest(url, options, body, timeout = 60000) {
  return new Promise((resolve, reject) => {
    const urlObj = new URL(url);
    const isHttps = urlObj.protocol === 'https:';
    const lib = isHttps ? https : http;
    
    const disguiseHeaders = buildDisguiseHeaders();
    const reqOptions = {
      hostname: urlObj.hostname,
      port: urlObj.port || (isHttps ? 443 : 80),
      path: urlObj.pathname + urlObj.search,
      method: options.method || 'GET',
      headers: { ...disguiseHeaders, ...options.headers },
      timeout
    };

    const req = lib.request(reqOptions, (res) => {
      let data = '';
      res.on('data', chunk => data += chunk);
      res.on('end', () => {
        resolve({ status: res.statusCode, headers: res.headers, body: data });
      });
    });

    req.on('error', reject);
    req.on('timeout', () => { req.destroy(); reject(new Error('Request timeout')); });
    
    if (body) req.write(body);
    req.end();
  });
}

async function fetchModels(baseUrl, token) {
  try {
    log.debug('Fetching models from:', baseUrl);
    const url = buildUrl(baseUrl, '/models');
    const resp = await makeRequest(url, {
      method: 'GET',
      headers: { 'Authorization': 'Bearer ' + token }
    });
    const data = JSON.parse(resp.body);
    if (resp.status !== 200) {
      throw new Error(data?.error?.message || 'HTTP ' + resp.status);
    }
    const ids = (Array.isArray(data?.data) ? data.data : [])
      .map(x => String(x.id || '').trim())
      .filter(Boolean)
      .sort();
    log.debug('Fetched', ids.length, 'models');
    return ids;
  } catch (e) {
    log.error('Failed to fetch models:', e.message);
    throw new Error('加载模型失败: ' + e.message);
  }
}

async function requestOnce(config, signal) {
  const { prompt, tag, marker } = buildUserPrompt(config.usr, config.randOn, config.markerLen);
  
  const body = {
    model: config.model,
    messages: [
      ...(config.sys ? [{ role: 'system', content: config.sys }] : []),
      { role: 'user', content: prompt }
    ],
    max_tokens: config.max,
    temperature: config.temp,
    stream: false
  };

  const url = buildUrl(config.base, '/chat/completions');
  const resp = await makeRequest(url, {
    method: 'POST',
    headers: {
      'Authorization': 'Bearer ' + config.token,
      'Content-Type': 'application/json'
    }
  }, JSON.stringify(body), config.timeout);

  if (signal?.aborted) throw new Error('Aborted');

  // Check HTTP status first
  if (resp.status !== 200) {
    let errorMsg = 'HTTP ' + resp.status;
    try {
      const errorData = JSON.parse(resp.body);
      errorMsg = errorData?.error?.message || errorMsg;
    } catch (e) {
      // Response body is not valid JSON, use raw body
      if (resp.body) errorMsg += ': ' + resp.body.slice(0, 200);
    }
    throw new Error(errorMsg);
  }

  // Parse successful response
  let data;
  try {
    data = JSON.parse(resp.body);
  } catch (e) {
    throw new Error('响应解析失败: ' + resp.body.slice(0, 200));
  }

  const content = data?.choices?.[0]?.message?.content || '';
  const usage = usageWithFallback(data?.usage, prompt, config.sys, content);

  return { content, usage, finish_reason: data?.choices?.[0]?.finish_reason, tag, marker, prompt };
}

async function streamRequest(config, onChunk, signal) {
  const { prompt, tag, marker } = buildUserPrompt(config.usr, config.randOn, config.markerLen);
  
  const body = {
    model: config.model,
    messages: [
      ...(config.sys ? [{ role: 'system', content: config.sys }] : []),
      { role: 'user', content: prompt }
    ],
    max_tokens: config.max,
    temperature: config.temp,
    stream: true,
    stream_options: { include_usage: true }
  };

  const url = buildUrl(config.base, '/chat/completions');
  const urlObj = new URL(url);
  const isHttps = urlObj.protocol === 'https:';
  const lib = isHttps ? https : http;

  return new Promise((resolve, reject) => {
    if (signal?.aborted) {
      reject(new Error('Aborted'));
      return;
    }

    const disguiseHeaders = buildDisguiseHeaders();
    const reqOptions = {
      hostname: urlObj.hostname,
      port: urlObj.port || (isHttps ? 443 : 80),
      path: urlObj.pathname,
      method: 'POST',
      headers: {
        ...disguiseHeaders,
        'Authorization': 'Bearer ' + config.token,
        'Content-Type': 'application/json'
      },
      timeout: config.timeout || 60000
    };

    let fullContent = '';
    let usage = { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 };
    let finishReason = null;
    let aborted = false;
    let abortHandler = null; // Define early for cleanup

    const cleanup = () => {
      if (abortHandler && signal) {
        signal.removeEventListener('abort', abortHandler);
      }
    };

    const req = lib.request(reqOptions, (res) => {
      let buffer = '';
      
      // Check HTTP status code for non-streaming errors
      if (res.statusCode !== 200) {
        res.on('data', chunk => buffer += chunk.toString());
        res.on('end', () => {
          cleanup();
          try {
            const data = JSON.parse(buffer);
            reject(new Error(data?.error?.message || 'HTTP ' + res.statusCode));
          } catch (e) {
            reject(new Error('HTTP ' + res.statusCode + ': ' + buffer.slice(0, 200)));
          }
        });
        return;
      }
      
      res.on('data', chunk => {
        if (aborted) { req.destroy(); return; }
        
        buffer += chunk.toString();
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
          const trimmed = line.trim();
          if (!trimmed || !trimmed.startsWith('data:')) continue;
          
          const payload = trimmed.slice(5).trim();
          if (payload === '[DONE]') continue;

          try {
            const json = JSON.parse(payload);
            if (json.error) throw new Error(json.error.message);
            
            const delta = json.choices?.[0]?.delta?.content;
            if (delta) {
              fullContent += delta;
              if (onChunk) onChunk(delta, fullContent);
            }
            
            if (json.usage) usage = json.usage;
            if (json.choices?.[0]?.finish_reason) {
              finishReason = json.choices[0].finish_reason;
            }
          } catch (e) {
            if (e.message && !e.message.includes('JSON')) {
              req.destroy();
              cleanup();
              reject(e);
              return;
            }
          }
        }
      });

      res.on('end', () => {
        cleanup();
        const finalUsage = usageWithFallback(usage, prompt, config.sys, fullContent);
        resolve({
          content: fullContent,
          usage: finalUsage,
          finish_reason: finishReason,
          tag,
          marker,
          prompt
        });
      });
    });

    req.on('error', (err) => {
      cleanup();
      if (aborted) reject(new Error('Aborted'));
      else reject(err);
    });
    
    req.on('timeout', () => {
      cleanup();
      aborted = true;
      req.destroy();
      reject(new Error('请求超时'));
    });
    
    req.write(JSON.stringify(body));
    req.end();

    // Set up abort handler
    if (signal) {
      abortHandler = () => {
        aborted = true;
        req.destroy();
        reject(new Error('Aborted'));
      };
      signal.addEventListener('abort', abortHandler);
    }
  });
}

class TaskExecutor {
  constructor(taskId, config, onUpdate, savedProgress = null) {
    this.taskId = taskId;
    this.config = config;
    this.onUpdate = onUpdate;
    
    this.running = false;
    this.stopped = false;
    this.paused = false;
    this.controller = new AbortController();
    
    // Update throttling - limit notification frequency
    this.lastNotifyTime = 0;
    this.notifyTimer = null;
    this.notifyInterval = 500; // ms between notifications (increased for high concurrency)
    
    // Load from saved progress or initialize fresh
    if (savedProgress) {
      this.stats = savedProgress.stats || {
        total: config.loop * config.threads,
        completed: 0,
        success: 0,
        failed: 0,
        aborted: 0,
        promptTokens: 0,
        completionTokens: 0,
        totalTokens: 0
      };
      this.threadProgress = savedProgress.threadProgress || {};
      this.markerLen = savedProgress.markerLen || config.markerLen || 24;
      this.ratio = savedProgress.ratio || { live: 0, target: config.ratioTarget || 1, markerLen: this.markerLen };
      this.elapsedBefore = savedProgress.elapsedBefore || 0;
      // Resume from paused state if saved
      if (savedProgress.paused) {
        this.paused = true;
        log.info('Resuming from paused state:', taskId);
      }
    } else {
      this.stats = {
        total: config.loop * config.threads,
        completed: 0,
        success: 0,
        failed: 0,
        aborted: 0,
        promptTokens: 0,
        completionTokens: 0,
        totalTokens: 0
      };
      this.threadProgress = {};
      this.markerLen = config.markerLen || 24;
      this.ratio = { live: 0, target: config.ratioTarget || 1, markerLen: this.markerLen };
      this.elapsedBefore = 0;
    }

    this.startTime = null;
    this.threadLogs = {};
    
    log.debug('TaskExecutor created:', taskId, 'threads:', config.threads, 'loop:', config.loop, 'resumed:', !!savedProgress);
  }

  async runWorker(threadId) {
    const startLoop = (this.threadProgress[threadId] || 0) + 1;
    
    for (let i = startLoop; i <= this.config.loop; i++) {
      // Check for pause
      while (this.paused && !this.stopped) {
        await new Promise(resolve => setTimeout(resolve, 100));
      }
      
      if (this.stopped) {
        log.debug('Thread', threadId, 'stopped at loop', i);
        break;
      }

      const threadLog = {
        thread: threadId,
        loop: i,
        status: 'running',
        startTime: Date.now(),
        message: ''
      };

      this.threadLogs[threadId] = threadLog;
      this.threadProgress[threadId] = i;
      this.notifyUpdate();

      try {
        const result = this.config.streamOn
          ? await streamRequest(
              { ...this.config, markerLen: this.markerLen },
              (delta, full) => {
                threadLog.status = 'streaming';
                threadLog.content = full;
                // notifyUpdate already throttled, so this is fine
                this.notifyUpdate();
              },
              this.controller.signal
            )
          : await requestOnce(
              { ...this.config, markerLen: this.markerLen },
              this.controller.signal
            );

        this.stats.completed++;
        this.stats.success++;
        this.stats.promptTokens += result.usage.prompt_tokens;
        this.stats.completionTokens += result.usage.completion_tokens;
        this.stats.totalTokens += result.usage.total_tokens;

        log.debug('Thread', threadId, 'loop', i, 'success, tokens:', result.usage.total_tokens);

        // Check token limit
        if (this.stats.totalTokens >= this.config.maxTokens) {
          this.stopped = true;
          threadLog.status = 'limit_reached';
          threadLog.message = '达到 tokens 上限';
          log.info('Task', this.taskId, 'reached token limit:', this.stats.totalTokens);
        } else {
          threadLog.status = 'success';
          threadLog.content = result.content?.slice(-500);
        }

        // Adjust marker for ratio
        const liveRatio = this.stats.completionTokens > 0
          ? this.stats.promptTokens / this.stats.completionTokens
          : 0;
        this.ratio.live = liveRatio;

        if (this.config.randOn) {
          const diff = liveRatio - this.ratio.target;
          if (Math.abs(diff) >= 0.02) {
            const step = Math.max(1, Math.round(Math.abs(diff) * 50));
            if (diff < 0) this.markerLen = Math.min(4000, this.markerLen + step);
            else this.markerLen = Math.max(0, this.markerLen - step);
          }
          this.ratio.markerLen = this.markerLen;
        }

      } catch (e) {
        const isAborted = e.message === 'Aborted' || e.name === 'AbortError';
        
        if (isAborted) {
          if (this.paused) {
            threadLog.status = 'paused';
            threadLog.message = '已暂停，等待继续...';
            log.debug('Thread', threadId, 'loop', i, 'paused');
            this.notifyUpdate(true);
            while (this.paused && !this.stopped) {
              await new Promise(resolve => setTimeout(resolve, 100));
            }
            if (!this.stopped) {
              i--;
              threadLog.status = 'running';
              continue;
            }
          }
          if (this.stopped) {
            this.stats.completed++;
            this.stats.aborted++;
            threadLog.status = 'aborted';
            log.debug('Thread', threadId, 'loop', i, 'aborted');
          }
        } else {
          this.stats.failed++;
          threadLog.status = 'error';
          threadLog.error = e.message || '请求失败';
          log.error('Thread', threadId, 'loop', i, 'error:', e.message);
        }
      }

      threadLog.endTime = Date.now();
      this.notifyUpdate(true); // Force update after each request completes
      
      // Wait between loops (not after the last loop)
      if (i < this.config.loop && this.config.waitBetween > 0 && !this.stopped && !this.paused) {
        await new Promise(resolve => setTimeout(resolve, this.config.waitBetween));
      }
    }
    
    // Clear progress when thread completes all loops
    if (!this.stopped && !this.paused) {
      delete this.threadProgress[threadId];
    }
  }

  async start() {
    log.info('TaskExecutor starting:', this.taskId, 'threads:', this.config.threads, 'loops:', this.config.loop, 'paused:', this.paused);
    this.running = true;
    this.stopped = false;
    
    // Always ensure controller exists
    if (!this.controller) {
      this.controller = new AbortController();
    }
    
    // Only reset startTime if not paused (will be set by resume())
    if (!this.paused) {
      this.startTime = Date.now();
      // Initialize thread logs for fresh start
      for (let i = 1; i <= this.config.threads; i++) {
        this.threadLogs[i] = { thread: i, status: 'waiting', message: '等待开始...' };
      }
    } else {
      // Resuming from paused state - restore thread logs
      for (let i = 1; i <= this.config.threads; i++) {
        if (!this.threadLogs[i]) {
          this.threadLogs[i] = { thread: i, status: 'paused', message: '等待继续...' };
        } else {
          this.threadLogs[i].status = 'paused';
          this.threadLogs[i].message = '等待继续...';
        }
      }
    }
    
    this.notifyUpdate();

    // Run workers in parallel
    const workers = [];
    for (let i = 1; i <= this.config.threads; i++) {
      workers.push(this.runWorker(i));
    }

    await Promise.all(workers);

    this.running = false;
    log.info('TaskExecutor finished:', this.taskId, 'success:', this.stats.success, 'failed:', this.stats.failed, 'tokens:', this.stats.totalTokens);
    this.notifyUpdate(true); // Force final update
    
    // Clean up callback reference to allow garbage collection
    this.onUpdate = null;
  }

  pause() {
    if (!this.running || this.paused) return;
    log.info('TaskExecutor pausing:', this.taskId);
    this.paused = true;
    if (this.startTime) {
      this.elapsedBefore += Date.now() - this.startTime;
      this.startTime = null;
    }
    // Abort current request to pause immediately
    this.controller.abort();
    this.notifyUpdate();
  }

  resume() {
    if (!this.running || !this.paused) return;
    log.info('TaskExecutor resuming:', this.taskId);
    this.paused = false;
    this.controller = new AbortController();
    this.startTime = Date.now();
    this.notifyUpdate();
  }

  stop() {
    log.info('TaskExecutor stopping:', this.taskId);
    this.stopped = true;
    this.paused = false;
    this.controller.abort();
  }

  getStatus() {
    let elapsed = this.elapsedBefore;
    if (this.startTime && !this.paused) {
      elapsed += Date.now() - this.startTime;
    }
    return {
      taskId: this.taskId,
      running: this.running,
      stopped: this.stopped,
      paused: this.paused,
      stats: this.stats,
      ratio: this.ratio,
      elapsed: elapsed,
      elapsedBefore: this.elapsedBefore,
      threadLogs: { ...this.threadLogs },
      threadProgress: { ...this.threadProgress },
      markerLen: this.markerLen
    };
  }

  getProgress() {
    let elapsed = this.elapsedBefore;
    if (this.startTime && !this.paused) {
      elapsed += Date.now() - this.startTime;
    }
    return {
      stats: this.stats,
      threadProgress: this.threadProgress,
      ratio: this.ratio,
      markerLen: this.markerLen,
      elapsedBefore: elapsed
    };
  }

  notifyUpdate(force = false) {
    // Throttle updates to reduce API load
    // force = true means immediate update (for errors, completion, etc.)
    const now = Date.now();
    const timeSinceLastNotify = now - this.lastNotifyTime;
    
    if (force || timeSinceLastNotify >= this.notifyInterval) {
      // Enough time passed or forced update, notify immediately
      if (this.notifyTimer) {
        clearTimeout(this.notifyTimer);
        this.notifyTimer = null;
      }
      this.lastNotifyTime = now;
      if (this.onUpdate) {
        this.onUpdate(this.getStatus());
      }
    } else if (!this.notifyTimer) {
      // Schedule a delayed notification - only one timer at a time
      this.notifyTimer = setTimeout(() => {
        this.notifyTimer = null;
        this.lastNotifyTime = Date.now();
        if (this.onUpdate) {
          this.onUpdate(this.getStatus());
        }
      }, this.notifyInterval - timeSinceLastNotify);
    }
    // If timer already exists, the scheduled notification will include latest state
  }
}

class HttpTaskExecutor {
  constructor(taskId, config, onUpdate, savedProgress = null) {
    this.taskId = taskId;
    this.config = config;
    this.onUpdate = onUpdate;
    
    this.running = false;
    this.stopped = false;
    this.paused = false;
    this.controller = new AbortController();
    
    this.lastNotifyTime = 0;
    this.notifyTimer = null;
    this.notifyInterval = 500;
    
    if (savedProgress) {
      this.stats = savedProgress.stats || {
        total: config.loop * config.threads,
        completed: 0,
        success: 0,
        failed: 0,
        aborted: 0,
        totalRequests: 0
      };
      this.threadProgress = savedProgress.threadProgress || {};
      this.elapsedBefore = savedProgress.elapsedBefore || 0;
      if (savedProgress.paused) {
        this.paused = true;
        log.info('Resuming HTTP task from paused state:', taskId);
      }
    } else {
      this.stats = {
        total: config.loop * config.threads,
        completed: 0,
        success: 0,
        failed: 0,
        aborted: 0,
        totalRequests: 0
      };
      this.threadProgress = {};
      this.elapsedBefore = 0;
    }

    this.startTime = null;
    this.threadLogs = {};
    
    log.debug('HttpTaskExecutor created:', taskId, 'threads:', config.threads, 'loop:', config.loop);
  }

  async executeHttpRequest(threadId, loopNum) {
    const { url, method, headers: headersStr, body: bodyStr, timeout } = this.config;
    
    let headers = {};
    try {
      if (headersStr) {
        headers = JSON.parse(headersStr);
      }
    } catch (e) {
      throw new Error('Headers JSON格式错误: ' + e.message);
    }
    
    let body = null;
    if (method.toUpperCase() !== 'GET' && bodyStr) {
      body = bodyStr;
      if (!headers['Content-Type']) {
        try {
          JSON.parse(bodyStr);
          headers['Content-Type'] = 'application/json';
        } catch {
          headers['Content-Type'] = 'application/x-www-form-urlencoded';
        }
      }
    }

    const urlObj = new URL(url);
    const isHttps = urlObj.protocol === 'https:';
    const lib = isHttps ? https : http;

    return new Promise((resolve, reject) => {
      const reqOptions = {
        hostname: urlObj.hostname,
        port: urlObj.port || (isHttps ? 443 : 80),
        path: urlObj.pathname + urlObj.search,
        method: method.toUpperCase(),
        headers: { ...headers },
        timeout: timeout || 60000
      };

      const req = lib.request(reqOptions, (res) => {
        let data = '';
        res.on('data', chunk => data += chunk);
        res.on('end', () => {
          resolve({
            statusCode: res.statusCode,
            headers: res.headers,
            body: data,
            success: res.statusCode >= 200 && res.statusCode < 300
          });
        });
      });

      req.on('error', err => reject(err));
      req.on('timeout', () => {
        req.destroy();
        reject(new Error('请求超时'));
      });
      
      if (body) req.write(body);
      req.end();
    });
  }

  async runWorker(threadId) {
    const startLoop = (this.threadProgress[threadId] || 0) + 1;
    
    for (let i = startLoop; i <= this.config.loop; i++) {
      while (this.paused && !this.stopped) {
        await new Promise(resolve => setTimeout(resolve, 100));
      }
      
      if (this.stopped) {
        break;
      }

      const threadLog = {
        thread: threadId,
        loop: i,
        status: 'running',
        startTime: Date.now(),
        message: '',
        responseData: null,
        errorCode: null,
        errorMessage: null
      };

      this.threadLogs[threadId] = threadLog;
      this.threadProgress[threadId] = i;
      this.notifyUpdate();

      try {
        const result = await this.executeHttpRequest(threadId, i);
        
        this.stats.completed++;
        this.stats.totalRequests++;
        
        if (result.success) {
          this.stats.success++;
          threadLog.status = 'success';
          threadLog.responseData = result.body;
          threadLog.message = `HTTP ${result.statusCode}`;
        } else {
          this.stats.failed++;
          threadLog.status = 'error';
          threadLog.errorCode = result.statusCode;
          threadLog.errorMessage = result.body.slice(0, 500);
          threadLog.message = `HTTP ${result.statusCode}`;
        }

      } catch (e) {
        this.stats.completed++;
        this.stats.failed++;
        threadLog.status = 'error';
        threadLog.errorCode = 'ERR';
        threadLog.errorMessage = e.message;
        threadLog.message = '请求失败: ' + e.message;
        log.error('Thread', threadId, 'loop', i, 'error:', e.message);
      }

      threadLog.endTime = Date.now();
      this.notifyUpdate(true);
      
      if (i < this.config.loop && this.config.waitBetween > 0 && !this.stopped && !this.paused) {
        await new Promise(resolve => setTimeout(resolve, this.config.waitBetween));
      }
    }
    
    if (!this.stopped && !this.paused) {
      delete this.threadProgress[threadId];
    }
  }

  async start() {
    log.info('HttpTaskExecutor starting:', this.taskId);
    this.running = true;
    this.stopped = false;
    
    if (!this.controller) {
      this.controller = new AbortController();
    }
    
    if (!this.paused) {
      this.startTime = Date.now();
      for (let i = 1; i <= this.config.threads; i++) {
        this.threadLogs[i] = { thread: i, status: 'waiting', message: '等待开始...' };
      }
    } else {
      for (let i = 1; i <= this.config.threads; i++) {
        if (!this.threadLogs[i]) {
          this.threadLogs[i] = { thread: i, status: 'paused', message: '等待继续...' };
        } else {
          this.threadLogs[i].status = 'paused';
          this.threadLogs[i].message = '等待继续...';
        }
      }
    }
    
    this.notifyUpdate();

    const workers = [];
    for (let i = 1; i <= this.config.threads; i++) {
      workers.push(this.runWorker(i));
    }

    await Promise.all(workers);

    this.running = false;
    log.info('HttpTaskExecutor finished:', this.taskId, 'success:', this.stats.success, 'failed:', this.stats.failed);
    this.notifyUpdate(true);
    this.onUpdate = null;
  }

  pause() {
    if (!this.running || this.paused) return;
    log.info('HttpTaskExecutor pausing:', this.taskId);
    this.paused = true;
    if (this.startTime) {
      this.elapsedBefore += Date.now() - this.startTime;
      this.startTime = null;
    }
    this.controller.abort();
    this.notifyUpdate();
  }

  resume() {
    if (!this.running || !this.paused) return;
    log.info('HttpTaskExecutor resuming:', this.taskId);
    this.paused = false;
    this.controller = new AbortController();
    this.startTime = Date.now();
    this.notifyUpdate();
  }

  stop() {
    log.info('HttpTaskExecutor stopping:', this.taskId);
    this.stopped = true;
    this.paused = false;
    this.controller.abort();
  }

  getStatus() {
    let elapsed = this.elapsedBefore;
    if (this.startTime && !this.paused) {
      elapsed += Date.now() - this.startTime;
    }
    return {
      taskId: this.taskId,
      running: this.running,
      stopped: this.stopped,
      paused: this.paused,
      stats: this.stats,
      elapsed: elapsed,
      elapsedBefore: this.elapsedBefore,
      threadLogs: { ...this.threadLogs },
      threadProgress: { ...this.threadProgress }
    };
  }

  getProgress() {
    let elapsed = this.elapsedBefore;
    if (this.startTime && !this.paused) {
      elapsed += Date.now() - this.startTime;
    }
    return {
      stats: this.stats,
      threadProgress: this.threadProgress,
      elapsedBefore: elapsed
    };
  }

  notifyUpdate(force = false) {
    const now = Date.now();
    const timeSinceLastNotify = now - this.lastNotifyTime;
    
    if (force || timeSinceLastNotify >= this.notifyInterval) {
      if (this.notifyTimer) {
        clearTimeout(this.notifyTimer);
        this.notifyTimer = null;
      }
      this.lastNotifyTime = now;
      if (this.onUpdate) {
        this.onUpdate(this.getStatus());
      }
    } else if (!this.notifyTimer) {
      this.notifyTimer = setTimeout(() => {
        this.notifyTimer = null;
        this.lastNotifyTime = Date.now();
        if (this.onUpdate) {
          this.onUpdate(this.getStatus());
        }
      }, this.notifyInterval - timeSinceLastNotify);
    }
  }
}

module.exports = { TaskExecutor, HttpTaskExecutor, fetchModels };