api-server/test/m-ai-03-synthetic.e2e-spec.ts

import { Test, TestingModule } from '@nestjs/testing';
import { INestApplication } from '@nestjs/common';
import { JwtService } from '@nestjs/jwt';
import request from 'supertest';
import { AppModule } from '../src/app.module';
import { AiJobCreationService } from '../src/modules/ai-job/ai-job-creation.service';
import { JobDefinitionRegistry } from '../src/modules/ai-job/job-definition-registry';
import { AiJobLifecycleRepository } from '../src/modules/ai-job/ai-job-lifecycle.repository';
import { AiJobStateMachine } from '../src/modules/ai-job/ai-job-state-machine';

/**
 * M-AI-03 Synthetic E2E — 真实基础设施
 *
 * 需要: MySQL + Redis + BullMQ (通过 docker start mysql redis)
 * 环境变量: NODE_ENV=test, AI_JOB_SYNTHETIC_ENABLED=true
 */
import * as net from 'net';

const userId = 'synthetic-e2e-user';
const OLD_ENV = { ...process.env };

/** 检查 MySQL/Redis 是否可达 */
async function checkInfra(): Promise<boolean> {
  const dbUrl = process.env.DATABASE_URL || '';
  const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';

  // Parse MySQL host:port from DATABASE_URL (mysql://user:pass@host:port/db)
  const dbMatch = dbUrl.match(/@([^:]+):(\d+)/);
  const dbHost = dbMatch?.[1] || '127.0.0.1';
  const dbPort = parseInt(dbMatch?.[2] || '3306', 10);

  // Parse Redis host:port
  const redisMatch = redisUrl.match(/@?([^:]+):(\d+)/);
  const redisHost = redisMatch?.[1] || '127.0.0.1';
  const redisPort = parseInt(redisMatch?.[2] || '6379', 10);

  const checkPort = (host: string, port: number): Promise<boolean> =>
    new Promise((resolve) => {
      const sock = new net.Socket();
      sock.setTimeout(2000);
      sock.on('connect', () => { sock.destroy(); resolve(true); });
      sock.on('error', () => resolve(false));
      sock.on('timeout', () => { sock.destroy(); resolve(false); });
      sock.connect(port, host);
    });

  const [mysqlOk, redisOk] = await Promise.all([
    checkPort(dbHost, dbPort),
    checkPort(redisHost, redisPort),
  ]);
  return mysqlOk && redisOk;
}

describe('M-AI-03 Synthetic E2E (real infra)', () => {
  let app: INestApplication;
  let creationService: AiJobCreationService;
  let registry: JobDefinitionRegistry;
  let lifecycleRepo: AiJobLifecycleRepository;
  let sm: AiJobStateMachine;
  let jwtService: JwtService;
  let userToken: string;
  let infraAvailable = false;

  beforeAll(async () => {
    infraAvailable = await checkInfra();
    if (!infraAvailable) {
      console.warn('[M-AI-03 E2E] MySQL/Redis not available — skipping E2E tests. Run: docker start mysql redis');
      return;
    }

    process.env.NODE_ENV = 'test';
    process.env.AI_JOB_SYNTHETIC_ENABLED = 'true';
    process.env.JWT_SECRET = 'synthetic-e2e-secret';

    const module: TestingModule = await Test.createTestingModule({
      imports: [AppModule],
    }).compile();

    app = module.createNestApplication();
    app.setGlobalPrefix('api', { exclude: ['admin-api/(.*)', 'internal/(.*)'] });
    await app.init();

    creationService = module.get(AiJobCreationService);
    registry = module.get(JobDefinitionRegistry);
    lifecycleRepo = module.get(AiJobLifecycleRepository);
    sm = module.get(AiJobStateMachine);
    jwtService = module.get(JwtService);

    userToken = jwtService.sign({
      sub: userId, email: 'e2e@test.com', role: 'USER', type: 'user',
    });
  }, 30000);

  afterAll(async () => {
    process.env = OLD_ENV;
    if (app) await app.close();
  });

  // Guard: skip all tests if infra not available
  const itIfInfra = (name: string, fn: () => Promise<void>) => {
    it(name, async () => {
      if (!infraAvailable) {
        console.log(`  [SKIP] Infra unavailable: ${name}`);
        return;
      }
      await fn();
    });
  };

  // ═══════════════ 场景 1-2: 创建成功 ═══════════════
  itIfInfra('1. Synthetic Definition 已注册', () => {
    expect(registry.has('synthetic_job')).toBe(true);
  });

  itIfInfra('2. AiJobCreationService 创建 synthetic_job → status=queued', async () => {
    const job = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-1',
    });
    expect(job).toBeDefined();
    expect(job.jobType).toBe('synthetic_job');
    expect(job.lifecycleStatus).toBe('queued');
    // 验证元数据来自 Definition
    expect(job.queueName).toBe('ai-interactive');
  });

  // ═══════════════ 场景 3: 幂等创建 ═══════════════
  itIfInfra('3. 相同 idempotencyKey → 返回同一 Job', async () => {
    const idemKey = `e2e-idem-${Date.now()}`;
    const j1 = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-3',
      idempotencyKey: idemKey,
    });
    const j2 = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-3',
      idempotencyKey: idemKey,
    });
    expect(j2.id).toBe(j1.id);
  });

  // ═══════════════ 场景 4: 原子创建 ═══════════════
  itIfInfra('4. Job + Snapshot + Outbox 原子创建', async () => {
    const { PrismaClient } = require('@prisma/client');
    const prisma = new PrismaClient();
    const job = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-4',
      idempotencyKey: `e2e-atomic-${Date.now()}`,
    });
    // Verify snapshot exists
    const snap = await prisma.aiJobSnapshot.findUnique({ where: { jobId: job.id } });
    expect(snap).toBeDefined();
    expect(snap.contentHash).toBeTruthy();
    // Verify outbox exists
    const outbox = await prisma.outboxEvent.findFirst({
      where: { aggregateId: job.id },
    });
    expect(outbox).toBeDefined();
    expect(outbox.eventType).toBe('ai.job.enqueue');
    await prisma.$disconnect();
  });

  // ═══════════════ 场景 13-14: Cancel ═══════════════
  itIfInfra('5. queued Job → cancel → cancelled', async () => {
    const job = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-5',
    });
    const res = await request(app.getHttpServer())
      .post(`/api/ai/jobs/${job.id}/cancel`)
      .set('Authorization', `Bearer ${userToken}`)
      .expect(200);
    expect(res.body.status).toBe('cancelled');
  });

  // ═══════════════ API 层验证 ═══════════════
  itIfInfra('6. GET /api/ai/jobs/:id — JWT 保护 401', async () => {
    await request(app.getHttpServer()).get('/api/ai/jobs/any').expect(401);
  });

  itIfInfra('7. GET /api/ai/jobs/:id — 用户隔离', async () => {
    const otherToken = jwtService.sign({
      sub: 'other-user', email: 'o@t.com', role: 'USER', type: 'user',
    });
    const job = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-7',
    });
    // Own job → 200
    await request(app.getHttpServer())
      .get(`/api/ai/jobs/${job.id}`)
      .set('Authorization', `Bearer ${userToken}`)
      .expect(200);
    // Other user → 403
    await request(app.getHttpServer())
      .get(`/api/ai/jobs/${job.id}`)
      .set('Authorization', `Bearer ${otherToken}`)
      .expect(403);
  });

  itIfInfra('8. 公开响应不含敏感字段', async () => {
    const job = await creationService.createJob({
      userId, jobType: 'synthetic_job', triggerType: 'user_api',
      targetType: 'synthetic', targetId: 'test-8',
    });
    const res = await request(app.getHttpServer())
      .get(`/api/ai/jobs/${job.id}`)
      .set('Authorization', `Bearer ${userToken}`)
      .expect(200);
    expect(res.body).not.toHaveProperty('validatedOutput');
    expect(res.body).not.toHaveProperty('internalErrorMessage');
    expect(res.body).not.toHaveProperty('snapshot');
  });

  // ═══════════════ 场景 16: 未知 JobType ═══════════════
  itIfInfra('9. 未知 jobType → UnknownJobTypeError', async () => {
    await expect(
      creationService.createJob({
        userId, jobType: 'nonexistent_type', triggerType: 'user_api',
        targetType: 'x', targetId: 'x',
      }),
    ).rejects.toThrow();
  });

  // ═══════════════ 场景 19: 生产环境保护 ═══════════════
  itIfInfra('10. NODE_ENV=test + AI_JOB_SYNTHETIC_ENABLED=true → 模块正常', () => {
    expect(process.env.NODE_ENV).toBe('test');
    expect(process.env.AI_JOB_SYNTHETIC_ENABLED).toBe('true');
  });

  // ═══════════════ 状态机验证 ═══════════════
  itIfInfra('11. 合法状态迁移全部通过', () => {
    expect(() => sm.validate('queued', 'running')).not.toThrow();
    expect(() => sm.validate('queued', 'cancelled')).not.toThrow();
    expect(() => sm.validate('running', 'succeeded')).not.toThrow();
    expect(() => sm.validate('running', 'failed')).not.toThrow();
    expect(() => sm.validate('running', 'cancelled')).not.toThrow();
  });

  itIfInfra('12. succeeded/failed/cancelled 是终态', () => {
    expect(sm.isTerminal('succeeded')).toBe(true);
    expect(sm.isTerminal('failed')).toBe(true);
    expect(sm.isTerminal('cancelled')).toBe(true);
    expect(sm.isTerminal('queued')).toBe(false);
  });
});