api-server/test/helpers/integration-harness.ts

/**
 * Integration Test Harness
 * 启动真实 API + Worker 进程，通过真实 HTTP 和 BullMQ 执行测试。
 */

import { spawn, ChildProcess } from 'child_process';
import * as http from 'http';
import { Queue } from 'bullmq';

let redisConn: { host: string; port: number; password?: string } | null = null;

export function setRedisConfig(host: string, port: number, password?: string) {
  redisConn = { host, port, password: password || undefined };
}

/**
 * Enqueue a job directly to BullMQ Redis queue.
 * Worker process consumes from this queue.
 */
export async function enqueueJob(queueName: string, data: Record<string, unknown>, jobId?: string): Promise<string> {
  if (!redisConn) throw new Error('Redis config not set — call setRedisConfig() first');
  const queue = new Queue(queueName, { connection: redisConn });
  const job = await queue.add(queueName, data, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    ...(jobId ? { jobId } : {}),
  });
  await queue.close();
  return job.id as string;
}

let apiProc: ChildProcess | null = null;
let workerProc: ChildProcess | null = null;
let mockAIUrl = '';

const API_PORT = 3000;

// ══════════════════════════════════════════════════════════════════
// HTTP helpers
// ══════════════════════════════════════════════════════════════════

export function httpGet(path: string): Promise<{ status: number; data: any }> {
  return new Promise((resolve, reject) => {
    http.get(`http://127.0.0.1:${API_PORT}${path}`, (res) => {
      let body = '';
      res.on('data', (c) => (body += c));
      res.on('end', () => {
        try { resolve({ status: res.statusCode || 0, data: JSON.parse(body) }); }
        catch { resolve({ status: res.statusCode || 0, data: body }); }
      });
    }).on('error', reject);
  });
}

export function httpPost(path: string, body: unknown): Promise<{ status: number; data: any }> {
  return new Promise((resolve, reject) => {
    const payload = JSON.stringify(body);
    const req = http.request({
      hostname: '127.0.0.1', port: API_PORT, path, method: 'POST',
      headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload) },
    }, (res) => {
      let b = '';
      res.on('data', (c) => (b += c));
      res.on('end', () => {
        try { resolve({ status: res.statusCode || 0, data: JSON.parse(b) }); }
        catch { resolve({ status: res.statusCode || 0, data: b }); }
      });
    });
    req.on('error', reject);
    req.write(payload);
    req.end();
  });
}

// ══════════════════════════════════════════════════════════════════
// Process management
// ══════════════════════════════════════════════════════════════════

function envForProcess(baseEnv: NodeJS.ProcessEnv): NodeJS.ProcessEnv {
  return {
    ...baseEnv,
    NODE_ENV: 'test',
    PORT: String(API_PORT),
    DEEPSEEK_BASE_URL: mockAIUrl,
    AI_PROVIDER: 'deepseek',
    DEEPSEEK_API_KEY: 'sk-test',
    STORAGE_DRIVER: 'local',
    ENABLE_SWAGGER: 'false',
    // No Bull Board in test
  };
}

export async function startAPI(mockUrl: string, dbUrl: string, redisHost: string, redisPort: string, redisPw: string): Promise<number> {
  mockAIUrl = mockUrl;
  return new Promise((resolve, reject) => {
    const env = envForProcess(process.env);
    Object.assign(env, {
      DATABASE_URL: dbUrl,
      REDIS_HOST: redisHost, REDIS_PORT: redisPort,
      REDIS_PASSWORD: redisPw, REDIS_DB: '0',
    });

    apiProc = spawn('node', ['dist/src/main.js'], { env, stdio: ['ignore', 'pipe', 'pipe'] });

    let done = false;
    const onData = (chunk: Buffer) => {
      if (done) return;
      const msg = chunk.toString();
      // API ready signal
      if (msg.includes('Server running') || msg.includes('successfully started')) {
        done = true;
        resolve(apiProc!.pid!);
      }
    };
    apiProc.stdout!.on('data', onData);
    apiProc.stderr!.on('data', onData);
    setTimeout(() => { if (!done) { done = true; reject(new Error('API timeout')); } }, 30_000);
  });
}

export async function startWorker(dbUrl: string, redisHost: string, redisPort: string, redisPw: string, instanceId: string): Promise<number> {
  return new Promise((resolve, reject) => {
    const env = envForProcess(process.env);
    Object.assign(env, {
      DATABASE_URL: dbUrl, REDIS_HOST: redisHost, REDIS_PORT: redisPort,
      REDIS_PASSWORD: redisPw, REDIS_DB: '0',
      WORKER_INSTANCE_ID: instanceId,
    });

    workerProc = spawn('node', ['dist/src/worker.main.js'], { env, stdio: ['ignore', 'pipe', 'pipe'] });

    let done = false;
    const onData = (chunk: Buffer) => {
      if (done) return;
      const msg = chunk.toString();
      if (msg.includes('started — waiting for jobs')) {
        done = true;
        resolve(workerProc!.pid!);
      }
    };
    workerProc!.stdout!.on('data', onData);
    workerProc!.stderr!.on('data', onData);
    setTimeout(() => { if (!done) { done = true; reject(new Error('Worker timeout')); } }, 30_000);
  });
}

export function getApiPid(): number | undefined { return apiProc?.pid; }
export function getWorkerPid(): number | undefined { return workerProc?.pid; }

export async function stopWorker(signal: NodeJS.Signals = 'SIGTERM'): Promise<void> {
  if (!workerProc) return;
  return new Promise((resolve) => {
    workerProc!.on('exit', () => resolve());
    workerProc!.kill(signal);
    setTimeout(() => resolve(), 10_000);
  });
}

export async function stopAPI(): Promise<void> {
  if (!apiProc) return;
  return new Promise((resolve) => {
    apiProc!.on('exit', () => resolve());
    apiProc!.kill('SIGTERM');
    setTimeout(() => resolve(), 10_000);
  });
}

// ══════════════════════════════════════════════════════════════════
// DB helpers (direct PrismaClient, no NestJS DI, no mocks)
// ══════════════════════════════════════════════════════════════════

import { PrismaClient } from '@prisma/client';

let prisma: PrismaClient | null = null;

export function getPrisma(dbUrl: string): PrismaClient {
  if (!prisma) {
    prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
  }
  return prisma;
}

export async function dbExec(dbUrl: string, sql: string, params: any[] = []): Promise<void> {
  const p = getPrisma(dbUrl);
  await p.$executeRawUnsafe(sql, ...params);
}

export async function dbQuery<T = any>(dbUrl: string, sql: string, params: any[] = []): Promise<T[]> {
  const p = getPrisma(dbUrl);
  return (await p.$queryRawUnsafe(sql, ...params)) as T[];
}

export async function dbClose(): Promise<void> {
  if (prisma) { await prisma.$disconnect(); prisma = null; }
}

export async function setupFixtures(dbUrl: string, userId: string, kbId: string, kiId: string, sessionId: string): Promise<void> {
  await dbExec(dbUrl, `INSERT IGNORE INTO User (id, email, nickname, role, status, updatedAt) VALUES (?, ?, ?, ?, ?, NOW())`,
    [userId, `test-${Date.now()}@zhixi.app`, 'Integration Test', 'USER', 'active']);
  await dbExec(dbUrl, `INSERT IGNORE INTO KnowledgeBase (id, userId, title, updatedAt) VALUES (?, ?, ?, NOW())`,
    [kbId, userId, 'Test KB']);
  await dbExec(dbUrl, `INSERT IGNORE INTO KnowledgeItem (id, userId, knowledgeBaseId, itemType, title, content, status, updatedAt) VALUES (?, ?, ?, ?, ?, ?, ?, NOW())`,
    [kiId, userId, kbId, 'concept', 'Test Item', 'Test content.', 'active']);
  await dbExec(dbUrl, `INSERT IGNORE INTO LearningSession (id, userId, knowledgeBaseId, knowledgeItemId, mode, status, startedAt, updatedAt) VALUES (?, ?, ?, ?, ?, ?, NOW(), NOW())`,
    [sessionId, userId, kbId, kiId, 'active_recall', 'active']);
}

export async function cleanupFixtures(dbUrl: string, userId: string): Promise<void> {
  const tables = ['ReviewCard', 'FocusItem', 'AiAnalysisResult', 'AiAnalysisJob', 'LearningSession', 'KnowledgeItem', 'KnowledgeBase'];
  for (const t of tables) {
    await dbExec(dbUrl, `DELETE FROM \`${t}\` WHERE userId = ?`, [userId]).catch(() => {});
  }
  await dbExec(dbUrl, `DELETE FROM User WHERE id = ?`, [userId]).catch(() => {});
}