宜博 / lemonboss

/**
 * JobManager — 异步 Skill 创建/重训任务管理器
 *
 * 管理长时运行的 Skill 任务（Hunt→Digest→Evolve）的生命周期。
 * 支持磁盘持久化，服务器重启后自动恢复历史 Job 状态。
 */

const { randomUUID } = require('crypto');
const fs = require('fs');
const path = require('path');

// Job TTL: 完成后保留 7 天（持久化后可保留更久）
const JOB_TTL = 7 * 24 * 60 * 60 * 1000;
const CLEANUP_INTERVAL = 5 * 60 * 1000; // 每 5 分钟清理一次

class JobManager {
  /**
   * @param {object} [options]
   * @param {string} [options.storePath] - 持久化文件路径。如果提供则启用磁盘持久化。
   */
  constructor(options = {}) {
    /** @type {Map<string, Job>} */
    this.jobs = new Map();
    this.storePath = options.storePath || null;

    // 从磁盘恢复
    if (this.storePath) {
      this._loadFromDisk();
    }

    // 定期清理过期任务
    this._cleanupTimer = setInterval(() => this._cleanup(), CLEANUP_INTERVAL);
    this._cleanupTimer.unref(); // 不阻止进程退出
  }

  /**
   * 创建新任务
   * @param {object} request
   * @returns {Job}
   */
  createJob(request) {
    const jobId = `job-${Date.now()}-${randomUUID().substring(0, 8)}`;
    const now = new Date().toISOString();

    const job = {
      jobId,
      status: 'queued',
      domain: request.domain || '',
      scope: request.scope || '',
      description: request.description || '',
      keywords: request.keywords || [],
      skillName: request.skillName || null,
      progress: '任务已加入队列',
      result: null,
      error: null,
      webhookUrl: request.webhookUrl || null,
      apiKey: request.apiKey || null,
      createdAt: now,
      updatedAt: now,
      // 额外字段透传
      ...(request.type ? { type: request.type } : {}),
      ...(request.conversationId ? { conversationId: request.conversationId } : {}),
    };

    this.jobs.set(jobId, job);
    this._persist();
    console.log(`[JobManager] Job created: ${jobId}`);
    return job;
  }

  /**
   * 更新任务状态
   * @param {string} jobId
   * @param {object} updates
   */
  updateJob(jobId, updates) {
    const job = this.jobs.get(jobId);
    if (!job) return null;

    Object.assign(job, updates, { updatedAt: new Date().toISOString() });
    this._persist();

    // 如果任务完成或失败，触发 webhook
    if ((updates.status === 'completed' || updates.status === 'failed') && job.webhookUrl) {
      this._sendWebhook(job).catch(err => {
        console.warn(`[JobManager] Webhook failed for ${jobId}:`, err.message);
      });
    }

    return job;
  }

  /**
   * 获取任务信息
   * @param {string} jobId
   * @returns {Job|null}
   */
  getJob(jobId) {
    return this.jobs.get(jobId) || null;
  }

  /**
   * 获取公开安全的 Job 信息（隐藏 apiKey, webhookUrl）
   */
  getPublicJob(jobId) {
    const job = this.getJob(jobId);
    if (!job) return null;
    return JobManager.toPublicJob(job);
  }

  /**
   * 列出任务（可按 apiKey 过滤）
   * @param {object} [options]
   * @param {string} [options.apiKey]
   * @param {number} [options.limit=20]
   * @returns {Job[]}
   */
  listJobs({ apiKey, limit = 20 } = {}) {
    let jobs = Array.from(this.jobs.values());
    if (apiKey) {
      jobs = jobs.filter(j => j.apiKey === apiKey);
    }
    // 按创建时间倒序
    jobs.sort((a, b) => new Date(b.createdAt) - new Date(a.createdAt));
    return jobs.slice(0, limit).map(j => JobManager.toPublicJob(j));
  }

  /**
   * 将 Job 转为公开安全字段
   */
  static toPublicJob(job) {
    const { apiKey, webhookUrl, ...rest } = job;
    return rest;
  }

  // ══════════════════════════════════════
  // 磁盘持久化
  // ══════════════════════════════════════

  /**
   * 保存所有 Job 到磁盘
   * @private
   */
  _persist() {
    if (!this.storePath) return;
    try {
      const dir = path.dirname(this.storePath);
      if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });

      const data = [];
      for (const job of this.jobs.values()) {
        data.push(job);
      }
      fs.writeFileSync(this.storePath, JSON.stringify(data, null, 2), 'utf-8');
    } catch (err) {
      console.warn(`[JobManager] Persist error:`, err.message);
    }
  }

  /**
   * 从磁盘恢复 Job
   * @private
   */
  _loadFromDisk() {
    if (!this.storePath || !fs.existsSync(this.storePath)) return;
    try {
      const raw = fs.readFileSync(this.storePath, 'utf-8');
      const data = JSON.parse(raw);
      if (!Array.isArray(data)) return;

      let loaded = 0;
      for (const job of data) {
        if (job.jobId) {
          // 将 running/hunting/digesting/creating 状态标记为中断
          if (!['completed', 'failed', 'queued'].includes(job.status)) {
            job.status = 'failed';
            job.progress = '服务器重启，任务中断';
            job.error = 'Server restarted while job was running';
            job.updatedAt = new Date().toISOString();
          }
          this.jobs.set(job.jobId, job);
          loaded++;
        }
      }
      if (loaded > 0) {
        console.log(`[JobManager] Restored ${loaded} jobs from ${this.storePath}`);
      }
    } catch (err) {
      console.warn(`[JobManager] Load error:`, err.message);
    }
  }

  // ══════════════════════════════════════
  // Webhook & Cleanup
  // ══════════════════════════════════════

  /**
   * 发送 Webhook 回调
   * @param {Job} job
   * @private
   */
  async _sendWebhook(job) {
    if (!job.webhookUrl) return;

    try {
      const response = await fetch(job.webhookUrl, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          jobId: job.jobId,
          status: job.status,
          skillName: job.skillName,
          progress: job.progress,
          error: job.error,
          updatedAt: job.updatedAt,
        }),
        signal: AbortSignal.timeout(10000),
      });
      console.log(`[JobManager] Webhook sent for ${job.jobId}: ${response.status}`);
    } catch (err) {
      console.warn(`[JobManager] Webhook error for ${job.jobId}:`, err.message);
    }
  }

  /**
   * 清理过期任务
   * @private
   */
  _cleanup() {
    const now = Date.now();
    let cleaned = 0;
    for (const [id, job] of this.jobs) {
      if (job.status === 'completed' || job.status === 'failed') {
        const age = now - new Date(job.updatedAt).getTime();
        if (age > JOB_TTL) {
          this.jobs.delete(id);
          cleaned++;
        }
      }
    }
    if (cleaned > 0) {
      this._persist();
      console.log(`[JobManager] Cleaned ${cleaned} expired jobs`);
    }
  }

  /**
   * 关闭清理定时器
   */
  shutdown() {
    if (this._cleanupTimer) {
      clearInterval(this._cleanupTimer);
      this._cleanupTimer = null;
    }
  }
}

module.exports = { JobManager };