宜博 / pixelphone

/**
 * FileBackend - 文件存储后端
 *
 * 将日志写入文件系统，支持：
 * - 按会话分目录存储
 * - 按日期分文件夹
 * - 自动轮转（按大小）
 * - JSONL 格式
 */

const fs = require('fs').promises;
const path = require('path');
const { createWriteStream } = require('fs');
const { BaseBackend } = require('./BaseBackend');

// 获取项目根目录
const PROJECT_ROOT = path.resolve(__dirname, '../../../../../');

class FileBackend extends BaseBackend {
  constructor(options = {}) {
    super({ name: 'file', level: 'debug', ...options }); // 默认 debug 级别，记录所有日志

    // 默认使用项目根目录的 .logs/
    this.baseDir = options.baseDir || path.join(PROJECT_ROOT, '.logs');
    this.maxFileSize = options.maxFileSize || 100 * 1024 * 1024; // 100MB
    this.maxFiles = options.maxFiles || 10;
    this.bufferSize = options.bufferSize || 100; // 缓冲条目数

    this.buffers = new Map(); // filePath -> entries[]
    this.streams = new Map(); // filePath -> writeStream
    this.fileSizes = new Map(); // filePath -> currentSize
    this.flushTimers = new Map(); // filePath -> timer
  }

  async init() {
    // 确保基础目录存在
    await fs.mkdir(this.baseDir, { recursive: true });

    // 启动定期刷新定时器
    // unref() 防止此 timer 阻止 Node.js 进程退出
    this.flushInterval = setInterval(() => {
      this._flushAll();
    }, 1000);
    this.flushInterval.unref();
  }

  /**
   * 计算日志文件的绝对路径
   */
  _getLogPath(entry) {
    const sessionId = entry.sessionId || 'default';

    // 1. 确定目录：.logs/{sessionId}/
    // 使用 path.resolve 确保路径规范化，防止路径遍历攻击
    // 将 sessionId 中的非法字符替换为 _
    const safeSessionId = sessionId.replace(/[^a-zA-Z0-9\-_]/g, '_');
    const sessionDir = path.join(this.baseDir, safeSessionId);

    // 2. 确定文件名
    let filename = 'orchestrator.jsonl'; // 默认 System/Orchestrator 日志

    // 如果是 Agent 相关日志，且有明确的 agentName
    if (entry.agentName && entry.agentName !== 'orchestrator') {
      const agentName = entry.agentName.replace(/[^a-zA-Z0-9\-_]/g, '_');

      // 尝试从 metadata 获取 iteration 信息
      const iteration = entry.metadata?.iteration;

      if (iteration !== undefined) {
        // 格式: 01_agentName.jsonl
        const iterStr = iteration.toString().padStart(2, '0');
        filename = `${iterStr}_${agentName}.jsonl`;

        // 如果只有 index (比如 executeWithDependencies 里的并行调用)
        if (entry.metadata?.index !== undefined) {
          filename = `${iterStr}_${agentName}_${entry.metadata.index}.jsonl`;
        }

      } else {
        // 没有 iteration，直接用 agentName
        filename = `${agentName}.jsonl`;
      }
    }

    return path.join(sessionDir, filename);
  }

  async write(entry) {
    if (!this._shouldLog(entry)) return;

    // 获取目标文件路径
    const filePath = this._getLogPath(entry);

    // 初始化缓冲区 (Key 为文件路径)
    if (!this.buffers.has(filePath)) {
      this.buffers.set(filePath, []);
    }

    const buffer = this.buffers.get(filePath);
    buffer.push(this._formatEntry(entry));

    // 缓冲满时刷新
    if (buffer.length >= this.bufferSize) {
      await this._flush(filePath);
    } else {
      // 设置延迟刷新
      this._scheduleFlush(filePath);
    }
  }

  async query(filters = {}) {
    const { sessionId, startTime, endTime, category, agentName, limit = 100 } = filters;

    const results = [];

    try {
      // 如果没有指定 sessionId，搜索所有会话
      const sessionsToSearch = sessionId
        ? [sessionId]
        : await this._getAllSessionIds();

      // 这里可能会有性能问题，因为现在每个 Session 有多个文件
      // 简单起见，我们优先读取 orchestrator.jsonl，如果用户查特定 agent 再读特定文件
      // 但为了保证时间线，理想情况是读取该 Session 下所有文件并 Merge Sort
      // 本次重构先实现合并读取。

      for (const sid of sessionsToSearch) {
        const entries = await this._readSessionLogs(sid); // 读取全部文件

        for (const entry of entries) {
          // 应用过滤器
          if (startTime && new Date(entry.timestamp) < new Date(startTime)) continue;
          if (endTime && new Date(entry.timestamp) > new Date(endTime)) continue;
          if (category && entry.category !== category) continue;
          if (agentName && entry.agentName !== agentName) continue;

          results.push(entry);

          // 注意：这里 limit 是针对全局的，为了更精确，应该全部读完再 sort 再 slice
          // 暂不 break，除非为了性能
        }
      }

      // 按时间戳全量排序
      return results.sort((a, b) => new Date(a.timestamp) - new Date(b.timestamp)).slice(0, limit);

    } catch (err) {
      console.error('[FileBackend] Query error:', err);
      return [];
    }
  }

  async close() {
    // 清除定时器
    if (this.flushInterval) {
      clearInterval(this.flushInterval);
    }

    // 取消所有延迟刷新
    for (const timer of this.flushTimers.values()) {
      clearTimeout(timer);
    }
    this.flushTimers.clear();

    // 刷新所有缓冲区
    await this._flushAll();

    // 关闭所有流
    for (const [filePath, stream] of this.streams) {
      stream.end();
    }
    this.streams.clear();
  }

  /**
   * 刷新指定文件的缓冲区
   * @param {string} filePath - 完整文件路径
   */
  async _flush(filePath) {
    const buffer = this.buffers.get(filePath);
    if (!buffer || buffer.length === 0) return;

    // 清空缓冲区
    this.buffers.set(filePath, []);
    clearTimeout(this.flushTimers.get(filePath));
    this.flushTimers.delete(filePath);

    try {
      // 确保目录存在
      const dir = path.dirname(filePath);
      await fs.mkdir(dir, { recursive: true });

      // 获取或创建写入流
      let stream = this.streams.get(filePath);
      let currentSize = this.fileSizes.get(filePath) || 0;

      // 检查是否需要轮转
      if (currentSize > this.maxFileSize) {
        await this._rotate(filePath);
        stream = null;
        currentSize = 0;
      }

      if (!stream) {
        stream = createWriteStream(filePath, { flags: 'a' });
        this.streams.set(filePath, stream);

        // 获取当前文件大小（仅首次打开时）
        if (!this.fileSizes.has(filePath)) {
          try {
            const stats = await fs.stat(filePath);
            currentSize = stats.size;
          } catch {
            currentSize = 0;
          }
        }
      }

      // 写入数据
      const data = buffer.map(e => JSON.stringify(e)).join('\n') + '\n';
      stream.write(data);

      // 更新大小
      currentSize += Buffer.byteLength(data, 'utf8');
      this.fileSizes.set(filePath, currentSize);

    } catch (err) {
      console.error(`[FileBackend] Failed to flush to ${filePath}:`, err);
    }
  }

  /**
   * 刷新所有缓冲区
   * @private
   */
  async _flushAll() {
    const promises = [];
    for (const filePath of this.buffers.keys()) {
      promises.push(this._flush(filePath));
    }
    await Promise.all(promises);
  }

  /**
   * 设置延迟刷新
   * @param {string} filePath
   */
  _scheduleFlush(filePath) {
    if (this.flushTimers.has(filePath)) return;

    const timer = setTimeout(() => {
      this._flush(filePath);
    }, 100); // 100ms 延迟
    timer.unref();

    this.flushTimers.set(filePath, timer);
  }

  /**
   * 轮转日志文件
   * @param {string} filePath
   */
  async _rotate(filePath) {
    const stream = this.streams.get(filePath);
    if (stream) {
      stream.end();
      this.streams.delete(filePath);
    }

    const rotatedFile = filePath.replace('.jsonl', `.jsonl.1`);

    try {
      await fs.rename(filePath, rotatedFile);
    } catch (err) {
      console.error(`[FileBackend] Failed to rotate log for ${filePath}:`, err);
    }

    this.fileSizes.set(filePath, 0);
  }

  /**
   * 获取所有会话ID
   * 扫描 .logs/ 下的所有子目录
   * @private
   */
  async _getAllSessionIds() {
    try {
      const items = await fs.readdir(this.baseDir, { withFileTypes: true });
      return items
        .filter(item => item.isDirectory())
        .map(item => item.name);
    } catch {
      return [];
    }
  }

  /**
   * 读取会话日志（合并目录下所有文件）
   * @private
   */
  async _readSessionLogs(sessionId) {
    const sessionDir = path.join(this.baseDir, sessionId);
    const entries = [];

    try {
      const files = await fs.readdir(sessionDir);
      const jsonlFiles = files.filter(f => f.endsWith('.jsonl') && !f.includes('.jsonl.'));

      for (const file of jsonlFiles) {
        const content = await fs.readFile(path.join(sessionDir, file), 'utf8');
        const lines = content.trim().split('\n').filter(line => line);

        for (const line of lines) {
          try {
            entries.push(JSON.parse(line));
          } catch {
            // 忽略
          }
        }
      }
    } catch {
      // 目录不存在
    }

    return entries;
  }
}

module.exports = { FileBackend };