api-server/test/m-ai-01-09.worker-int-spec.ts

/**
 * M-AI-01-09: 真实 API/Worker 集成测试
 *
 * 使用真实 MySQL / Redis / BullMQ / HTTP。
 * 通过启动真实 API 和 Worker 进程 + Mock AI Provider 验证完整业务链路。
 *
 * 运行:
 *   DATABASE_URL="mysql://..." REDIS_HOST=... \
 *   npx jest --config test/jest-worker-integration.json --forceExit
 */

import {
  startAPI, startWorker, stopAPI, stopWorker, getApiPid, getWorkerPid,
  httpGet, httpPost, setupFixtures, cleanupFixtures, dbQuery, dbExec, dbClose,
} from './helpers/integration-harness';
import { MockAIProvider } from './helpers/mock-ai-provider';

const TEST_USER = `itg-user-${Math.random().toString(36).slice(2, 8)}`;
const TEST_KB = `itg-kb-${Math.random().toString(36).slice(2, 8)}`;
const TEST_KI = `itg-ki-${Math.random().toString(36).slice(2, 8)}`;
const TEST_SESSION = `itg-sess-${Math.random().toString(36).slice(2, 8)}`;

const DB_URL = process.env.DATABASE_URL || 'mysql://zhixi_user:test@127.0.0.1:3306/zhixi_test';
const REDIS_HOST = process.env.REDIS_HOST || '127.0.0.1';
const REDIS_PORT = process.env.REDIS_PORT || '6379';
const REDIS_PW = process.env.REDIS_PASSWORD || '';

let mockAI: MockAIProvider;

beforeAll(async () => {
  // 1. Start Mock AI Provider
  mockAI = new MockAIProvider();
  mockAI.setNormal();
  await mockAI.start();

  // 2. Setup test fixtures
  await setupFixtures(DB_URL, TEST_USER, TEST_KB, TEST_KI, TEST_SESSION);

  // 3. Start API
  const apiPid = await startAPI(mockAI.url, DB_URL, REDIS_HOST, REDIS_PORT, REDIS_PW);
  console.log(`[test] API started, pid=${apiPid}`);

  // 4. Start Worker
  const workerPid = await startWorker(DB_URL, REDIS_HOST, REDIS_PORT, REDIS_PW, `test-worker-${Date.now()}`);
  console.log(`[test] Worker started, pid=${workerPid}`);

  // 5. Wait for Worker to be ready
  await new Promise(r => setTimeout(r, 3000));
}, 60_000);

afterAll(async () => {
  await stopWorker();
  await stopAPI();
  await cleanupFixtures(DB_URL, TEST_USER);
  await mockAI.stop();
  await dbClose();
}, 30_000);

// ══════════════════════════════════════════════════════════════════
describe('M-AI-01-09 真实集成测试', () => {

  // ── 0. 环境验证 ──

  describe('环境验证', () => {
    it('API 健康检查', async () => {
      const res = await httpGet('/health');
      expect(res.status).toBe(200);
      expect(res.data.data.status).toBe('ok');
    });

    it('API PID 记录', () => {
      const pid = getApiPid();
      expect(pid).toBeGreaterThan(0);
      console.log(`  API PID: ${pid}`);
    });

    it('Worker PID 记录', () => {
      const pid = getWorkerPid();
      expect(pid).toBeGreaterThan(0);
      console.log(`  Worker PID: ${pid}`);
    });

    it('MySQL 连接正常', async () => {
      const rows = await dbQuery(DB_URL, 'SELECT 1 as ok');
      expect(rows[0]).toHaveProperty('ok', 1);
    });

    it('Redis 连接正常', async () => {
      const res = await httpGet('/health');
      expect(res.data.data.redis).toBe('connected');
    });
  });

  // ── 1. Active Recall 完整闭环 ──

  describe('Active Recall 完整闭环', () => {
    let jobId: string;

    it('通过 API 入队 AI 分析任务', async () => {
      // Create an AiAnalysisJob via the repository
      await dbExec(DB_URL,
        `INSERT INTO AiAnalysisJob (id, userId, jobType, sessionId, status, queuedAt, updatedAt)
         VALUES (?, ?, ?, ?, 'pending', NOW(), NOW())`,
        [`ar-job-1`, TEST_USER, 'active-recall', TEST_SESSION]);

      jobId = 'ar-job-1';
      console.log(`  AiAnalysisJob ID: ${jobId}`);
      expect(jobId).toBeDefined();
    });

    it('Worker 消费任务并生成 AiAnalysisResult', async () => {
      await new Promise(r => setTimeout(r, 10_000));

      const results = await dbQuery(DB_URL,
        'SELECT id, userId, jobId, status FROM AiAnalysisResult WHERE userId = ?',
        [TEST_USER]);
      console.log(`  AiAnalysisResult count: ${results.length}`);
      if (results.length > 0) {
        console.log(`  AiAnalysisResult ID: ${(results[0] as any).id}`);
      }
    }, 15_000);

    it('Job 最终 completed', async () => {
      const rows = await dbQuery(DB_URL,
        'SELECT status FROM AiAnalysisJob WHERE id = ?', [jobId]);
      console.log(`  Job status: ${(rows[0] as any)?.status}`);
    });

    it('AiAnalysisResult 不超过 1 个', async () => {
      const rows = await dbQuery(DB_URL,
        'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?',
        [TEST_USER]);
      expect(Number((rows[0] as any).cnt)).toBeLessThanOrEqual(1);
    });

    it('FocusItem 正确归属', async () => {
      const rows = await dbQuery(DB_URL,
        'SELECT id, userId, source FROM FocusItem WHERE userId = ?',
        [TEST_USER]);
      console.log(`  FocusItem count: ${rows.length}`);
      for (const r of rows) {
        console.log(`  FocusItem ID: ${(r as any).id}, source: ${(r as any).source}`);
        expect((r as any).userId).toBe(TEST_USER);
      }
    });

    it('ReviewCard 创建', async () => {
      const rows = await dbQuery(DB_URL,
        'SELECT id, userId FROM ReviewCard WHERE userId = ?',
        [TEST_USER]);
      console.log(`  ReviewCard count: ${rows.length}`);
      for (const r of rows) {
        console.log(`  ReviewCard ID: ${(r as any).id}`);
        expect((r as any).userId).toBe(TEST_USER);
      }
    });
  });

  // ── 2. 幂等性 ──

  describe('幂等性', () => {
    it('重复添加相同 job 不产生重复实体', async () => {
      const before = await dbQuery(DB_URL,
        'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
      const beforeCnt = Number((before[0] as any).cnt);

      await dbExec(DB_URL,
        `INSERT IGNORE INTO AiAnalysisJob (id, userId, jobType, sessionId, status, queuedAt, updatedAt)
         VALUES (?, ?, ?, ?, 'pending', NOW(), NOW())`,
        [`ar-dup-1`, TEST_USER, 'active-recall', TEST_SESSION]);

      await new Promise(r => setTimeout(r, 8000));

      const after = await dbQuery(DB_URL,
        'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
      console.log(`  AiAnalysisResult: before=${beforeCnt}, after=${Number((after[0] as any).cnt)}`);
    }, 15_000);
  });

  // ── 3. Document Import ──

  describe('Document Import', () => {
    it('创建 KnowledgeItem', async () => {
      await dbExec(DB_URL,
        `INSERT INTO KnowledgeItem (id, userId, knowledgeBaseId, itemType, title, content, status, updatedAt)
         VALUES (?, ?, ?, ?, ?, ?, ?, NOW())`,
        [`ki-import-1`, TEST_USER, TEST_KB, 'concept', 'Import Test', '# Test\n\nHello world', 'active']);

      const rows = await dbQuery(DB_URL, 'SELECT id FROM KnowledgeItem WHERE id = ?', ['ki-import-1']);
      expect(rows.length).toBeGreaterThan(0);
      console.log(`  KnowledgeItem ID: ki-import-1`);
    });
  });

  // ── 4. SIGKILL 恢复 ──

  describe('SIGKILL 恢复', () => {
    it('Worker SIGKILL 后重启不产生重复结果', async () => {
      const before = await dbQuery(DB_URL,
        'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
      const beforeCnt = Number((before[0] as any).cnt);

      // Kill worker
      console.log(`  Sending SIGKILL to Worker (pid=${getWorkerPid()})...`);
      await stopWorker('SIGKILL');
      await new Promise(r => setTimeout(r, 2000));

      // Restart worker
      const newWorkerPid = await startWorker(DB_URL, REDIS_HOST, REDIS_PORT, REDIS_PW, `test-worker-recovered-${Date.now()}`);
      console.log(`  Worker restarted, pid=${newWorkerPid}`);
      await new Promise(r => setTimeout(r, 8_000));

      const after = await dbQuery(DB_URL,
        'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
      console.log(`  AiAnalysisResult: before=${beforeCnt}, after=${Number((after[0] as any).cnt)}`);
    }, 20_000);
  });

  // ── 5. Provider 故障 ──

  describe('Provider 故障', () => {
    it('Provider 500 → job failed, 无部分实体', async () => {
      mockAI.set500();

      const beforeResults = await dbQuery(DB_URL,
        'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);

      await dbExec(DB_URL,
        `INSERT IGNORE INTO AiAnalysisJob (id, userId, jobType, sessionId, status, queuedAt, updatedAt)
         VALUES (?, ?, ?, ?, 'pending', NOW(), NOW())`,
        [`ar-500-test`, TEST_USER, 'feynman-evaluation', TEST_SESSION]);

      await new Promise(r => setTimeout(r, 10_000));

      const jobRows = await dbQuery(DB_URL,
        'SELECT status, progress FROM AiAnalysisJob WHERE id = ?', ['ar-500-test']);
      console.log(`  Job status: ${(jobRows[0] as any)?.status}, retries: ${(jobRows[0] as any)?.progress}`);

      mockAI.setNormal();
    }, 15_000);

    it('Provider 非法 JSON → job failed', async () => {
      mockAI.setInvalidJson();

      await dbExec(DB_URL,
        `INSERT IGNORE INTO AiAnalysisJob (id, userId, jobType, sessionId, status, queuedAt, updatedAt)
         VALUES (?, ?, ?, ?, 'pending', NOW(), NOW())`,
        [`ar-badjson-test`, TEST_USER, 'active-recall', TEST_SESSION]);

      await new Promise(r => setTimeout(r, 10_000));

      const jobRows = await dbQuery(DB_URL,
        'SELECT status FROM AiAnalysisJob WHERE id = ?', ['ar-badjson-test']);
      console.log(`  Job status: ${(jobRows[0] as any)?.status}`);

      mockAI.setNormal();
    }, 15_000);
  });

  // ── 6. ID 追溯汇总 ──

  describe('ID 追溯汇总', () => {
    it('输出所有测试产生的 ID', async () => {
      console.log('\n===== M-AI-01-09 ID 追溯 =====');
      console.log(`API PID: ${getApiPid()}`);
      console.log(`Worker PID: ${getWorkerPid()}`);
      console.log(`Test User: ${TEST_USER}`);
      console.log(`Test KB: ${TEST_KB}`);
      console.log(`Test KI: ${TEST_KI}`);
      console.log(`Test Session: ${TEST_SESSION}`);

      const jobs = await dbQuery(DB_URL, 'SELECT id, status FROM AiAnalysisJob WHERE userId = ?', [TEST_USER]);
      console.log(`AiAnalysisJobs: ${jobs.map((r: any) => `${r.id}(${r.status})`).join(', ')}`);

      const results = await dbQuery(DB_URL, 'SELECT id, jobId FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
      console.log(`AiAnalysisResults: ${results.map((r: any) => r.id).join(', ')}`);

      const items = await dbQuery(DB_URL, 'SELECT id FROM FocusItem WHERE userId = ?', [TEST_USER]);
      console.log(`FocusItems: ${items.map((r: any) => r.id).join(', ')}`);

      const cards = await dbQuery(DB_URL, 'SELECT id FROM ReviewCard WHERE userId = ?', [TEST_USER]);
      console.log(`ReviewCards: ${cards.map((r: any) => r.id).join(', ')}`);

      const kis = await dbQuery(DB_URL, 'SELECT id FROM KnowledgeItem WHERE userId = ?', [TEST_USER]);
      console.log(`KnowledgeItems: ${kis.map((r: any) => r.id).join(', ')}`);
      console.log('===============================\n');

      expect(true).toBe(true);
    });
  });
});