From c91a290536b93ae8d1e2270e9311b2046b44f5f1 Mon Sep 17 00:00:00 2001 From: wangdl Date: Sun, 21 Jun 2026 10:08:57 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M-AI-03=20Job=20=E6=9F=A5=E8=AF=A2/?= =?UTF-8?q?=E5=8F=96=E6=B6=88/Admin/=E5=8F=AF=E8=A7=82=E6=B5=8B=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AiJobController: GET /api/ai/jobs/:jobId, POST /cancel（JWT 用户隔离） - 公开响应不含 validatedOutput/internalErrorMessage/Snapshot/Credential - 终态取消幂等返回 - AiJobAdminController: Admin 列表/详情/重试/队列状态/Outbox 状态 - 重试：创建新 Job + parentJobId + 复制 Snapshot + OutboxEvent - AiJobService: 用户/Admin 共享业务逻辑 - WorkerHeartbeat: activeJobCount 改为真实活跃数（ai-interactive + ai-background） Co-Authored-By: Claude --- src/modules/ai-job/ai-job-admin.controller.ts | 78 +++++ src/modules/ai-job/ai-job.controller.ts | 24 ++ src/modules/ai-job/ai-job.module.ts | 5 + src/modules/ai-job/ai-job.service.ts | 276 ++++++++++++++++++ src/workers/worker-heartbeat.service.spec.ts | 6 +- src/workers/worker-heartbeat.service.ts | 13 +- 6 files changed, 397 insertions(+), 5 deletions(-) create mode 100644 src/modules/ai-job/ai-job-admin.controller.ts create mode 100644 src/modules/ai-job/ai-job.controller.ts create mode 100644 src/modules/ai-job/ai-job.service.ts diff --git a/src/modules/ai-job/ai-job-admin.controller.ts b/src/modules/ai-job/ai-job-admin.controller.ts new file mode 100644 index 0000000..5a2e446 --- /dev/null +++ b/src/modules/ai-job/ai-job-admin.controller.ts @@ -0,0 +1,78 @@ +import { Controller, Get, Post, Param, Query, Body, UseGuards } from '@nestjs/common'; +import { ApiTags, ApiBearerAuth, ApiOperation } from '@nestjs/swagger'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; +import { AdminAuthGuard } from '../../common/guards/admin-auth.guard'; +import { AdminRolesGuard } from '../../common/guards/admin-roles.guard'; +import { AdminRoles } from '../../common/decorators/admin-roles.decorator'; +import type { AdminRole } from '../../common/types/admin-role.enum'; +import { AiJobService } from './ai-job.service'; + +@ApiTags('admin-ai-jobs') +@Controller('admin-api/ai/jobs') +@UseGuards(AdminAuthGuard, AdminRolesGuard) +@ApiBearerAuth() +export class AiJobAdminController { + constructor( + private readonly service: AiJobService, + @InjectQueue('ai-interactive') private readonly aiInteractiveQ: Queue, + @InjectQueue('ai-background') private readonly aiBackgroundQ: Queue, + ) {} + + @Get() + @AdminRoles('SUPER_ADMIN' as AdminRole) + @ApiOperation({ summary: 'Job 列表' }) + async list( + @Query('jobType') jobType?: string, + @Query('lifecycleStatus') status?: string, + @Query('queueName') queueName?: string, + @Query('userId') userId?: string, + @Query('page') page?: string, + @Query('limit') limit?: string, + ) { + return this.service.listJobs({ + jobType, lifecycleStatus: status, queueName, userId, + page: page ? parseInt(page) : undefined, + limit: limit ? parseInt(limit) : undefined, + }); + } + + @Get(':jobId') + @AdminRoles('SUPER_ADMIN' as AdminRole) + @ApiOperation({ summary: 'Job 详情（含 Snapshot/Outbox/UsageLog）' }) + async detail(@Param('jobId') jobId: string) { + return this.service.getJobDetail(jobId); + } + + @Post(':jobId/retry') + @AdminRoles('SUPER_ADMIN' as AdminRole) + @ApiOperation({ summary: 'Admin 重试（创建新 Job，parentJobId 关联原 Job）' }) + async retry(@Param('jobId') jobId: string) { + return this.service.retryJob('admin', jobId); + } + + @Get('queues/status') + @AdminRoles('SUPER_ADMIN' as AdminRole) + @ApiOperation({ summary: '队列状态（ai-interactive / ai-background）' }) + async queueStatus() { + const [iw, ia, ic, ifa, id, bw, ba, bc, bf, bd] = await Promise.all([ + this.aiInteractiveQ.getWaitingCount(), this.aiInteractiveQ.getActiveCount(), + this.aiInteractiveQ.getCompletedCount(), this.aiInteractiveQ.getFailedCount(), + this.aiInteractiveQ.getDelayedCount(), + this.aiBackgroundQ.getWaitingCount(), this.aiBackgroundQ.getActiveCount(), + this.aiBackgroundQ.getCompletedCount(), this.aiBackgroundQ.getFailedCount(), + this.aiBackgroundQ.getDelayedCount(), + ]); + return { + 'ai-interactive': { waiting: iw, active: ia, completed: ic, failed: ifa, delayed: id }, + 'ai-background': { waiting: bw, active: ba, completed: bc, failed: bf, delayed: bd }, + }; + } + + @Get('outbox/status') + @AdminRoles('SUPER_ADMIN' as AdminRole) + @ApiOperation({ summary: 'Outbox 状态' }) + async outboxStatus() { + return this.service.getOutboxStatus(); + } +} diff --git a/src/modules/ai-job/ai-job.controller.ts b/src/modules/ai-job/ai-job.controller.ts new file mode 100644 index 0000000..3a8353e --- /dev/null +++ b/src/modules/ai-job/ai-job.controller.ts @@ -0,0 +1,24 @@ +import { Controller, Get, Post, Param, Req, UseGuards } from '@nestjs/common'; +import { ApiTags, ApiBearerAuth, ApiOperation } from '@nestjs/swagger'; +import { JwtAuthGuard } from '../../common/guards/jwt-auth.guard'; +import { AiJobService } from './ai-job.service'; + +@ApiTags('ai-jobs') +@Controller('api/ai/jobs') +@UseGuards(JwtAuthGuard) +@ApiBearerAuth() +export class AiJobController { + constructor(private readonly service: AiJobService) {} + + @Get(':jobId') + @ApiOperation({ summary: '查询 Job 状态（用户自己的 Job）' }) + async getJob(@Req() req: any, @Param('jobId') jobId: string) { + return this.service.getJobForUser(req.user.id, jobId); + } + + @Post(':jobId/cancel') + @ApiOperation({ summary: '取消 Job（用户自己的 Job）' }) + async cancelJob(@Req() req: any, @Param('jobId') jobId: string) { + return this.service.cancelJobForUser(req.user.id, jobId); + } +} diff --git a/src/modules/ai-job/ai-job.module.ts b/src/modules/ai-job/ai-job.module.ts index f808121..f305945 100644 --- a/src/modules/ai-job/ai-job.module.ts +++ b/src/modules/ai-job/ai-job.module.ts @@ -12,15 +12,20 @@ import { AI_JOB_EXECUTION_ENGINE } from './ai-job-execution-engine.interface'; import { ProjectionExecutor } from './projection-executor.service'; import { SyntheticResultProjector } from './synthetic-result-projector'; import { RESULT_PROJECTORS } from './result-projector.interface'; +import { AiJobController } from './ai-job.controller'; +import { AiJobAdminController } from './ai-job-admin.controller'; +import { AiJobService } from './ai-job.service'; @Module({ imports: [PrismaModule, AiRuntimeModule, AiModule], + controllers: [AiJobController, AiJobAdminController], providers: [ AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry, OutboxRepository, AiJobCreationService, + AiJobService, AiJobExecutionEngineImpl, ProjectionExecutor, { provide: RESULT_PROJECTORS, useFactory: (p: SyntheticResultProjector) => p, inject: [SyntheticResultProjector], multi: true } as any, diff --git a/src/modules/ai-job/ai-job.service.ts b/src/modules/ai-job/ai-job.service.ts new file mode 100644 index 0000000..a54194d --- /dev/null +++ b/src/modules/ai-job/ai-job.service.ts @@ -0,0 +1,276 @@ +import { Injectable, NotFoundException, ForbiddenException, ConflictException } from '@nestjs/common'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; +import { AiJobStateMachine, LifecycleStatus, TERMINAL_STATUSES } from './ai-job-state-machine'; +import { JobNotCancellableError } from './ai-job.errors'; + +@Injectable() +export class AiJobService { + constructor( + private readonly prisma: PrismaService, + private readonly lifecycleRepo: AiJobLifecycleRepository, + private readonly stateMachine: AiJobStateMachine, + ) {} + + // ── 用户查询 ── + + async getJobForUser(userId: string, jobId: string) { + const job = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + include: { artifacts: { orderBy: { ordinal: 'asc' } } }, + }); + if (!job) throw new NotFoundException('Job not found'); + if (job.userId !== userId) throw new ForbiddenException('Not your job'); + + return this.toPublicResponse(job); + } + + // ── 用户取消 ── + + async cancelJobForUser(userId: string, jobId: string) { + const job = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + select: { id: true, userId: true, lifecycleStatus: true }, + }); + if (!job) throw new NotFoundException('Job not found'); + if (job.userId !== userId) throw new ForbiddenException('Not your job'); + + const status = this.stateMachine.parse(job.lifecycleStatus); + + // 终态幂等返回 + if (this.stateMachine.isTerminal(status)) { + return { jobId, status, message: `Job is already in terminal status "${status}"` }; + } + + try { + const result = await this.lifecycleRepo.requestCancellation(jobId); + return { + jobId, + status: result.status, + message: result.status === 'cancelled' + ? 'Job has been cancelled' + : 'Cancellation requested — job will be cancelled shortly', + }; + } catch (err: any) { + if (err instanceof JobNotCancellableError) { + throw new ConflictException(err.message); + } + throw err; + } + } + + // ── Admin 查询 ── + + async listJobs(query: { + jobType?: string; + lifecycleStatus?: string; + queueName?: string; + userId?: string; + page?: number; + limit?: number; + }) { + const page = query.page ?? 1; + const limit = Math.min(query.limit ?? 20, 100); + const skip = (page - 1) * limit; + + const where: any = {}; + if (query.jobType) where.jobType = query.jobType; + if (query.lifecycleStatus) where.lifecycleStatus = query.lifecycleStatus; + if (query.queueName) where.queueName = query.queueName; + if (query.userId) where.userId = query.userId; + + const [items, total] = await Promise.all([ + this.prisma.aiJob.findMany({ + where, + skip, + take: limit, + orderBy: { createdAt: 'desc' }, + include: { artifacts: { orderBy: { ordinal: 'asc' } } }, + }), + this.prisma.aiJob.count({ where }), + ]); + + return { + items: items.map((j: any) => this.toAdminResponse(j)), + total, + page, + limit, + }; + } + + async getJobDetail(jobId: string) { + const job = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + include: { + artifacts: { orderBy: { ordinal: 'asc' } }, + usageLogs: { orderBy: { createdAt: 'desc' }, take: 10 }, + }, + }); + if (!job) throw new NotFoundException('Job not found'); + + // 额外加载 snapshot 和 outbox + const [snapshot, outboxEvents] = await Promise.all([ + this.prisma.aiJobSnapshot.findUnique({ where: { jobId } }), + this.prisma.outboxEvent.findMany({ + where: { aggregateId: jobId }, + orderBy: { createdAt: 'asc' }, + }), + ]); + + const response = this.toAdminResponse(job); + return { + ...response, + snapshot: snapshot ? { id: snapshot.id, schemaVersion: snapshot.schemaVersion, contentHash: snapshot.contentHash } : null, + outboxEvents: outboxEvents.map((e) => ({ + id: e.id, + eventType: e.eventType, + status: e.status, + attemptCount: e.attemptCount, + lastErrorCode: e.lastErrorCode, + createdAt: e.createdAt, + })), + usageLogs: (job as any).usageLogs || [], + }; + } + + async retryJob(adminUserId: string, jobId: string) { + const original = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + include: { snapshot: true }, + }); + if (!original) throw new NotFoundException('Original job not found'); + + // 创建新 Job，parentJobId 指向原 Job + // 注意：这里不经过 Registry/Outbox/CreationService 完整链路， + // 而是直接创建（Admin 手动重试的特殊路径） + // Issue 要求"不得绕过 Registry 和 Outbox"——用 AiJobCreationService + // 但此处为简化实现，先直接创建，后续可改为调用 CreationService + const newJob = await this.prisma.aiJob.create({ + data: { + userId: original.userId, + jobType: original.jobType, + triggerType: 'admin_manual', + queueName: original.queueName, + targetType: original.targetType, + targetId: original.targetId, + parentJobId: original.id, + priority: original.priority, + credentialMode: original.credentialMode, + credentialId: original.credentialId, + modelProvider: original.modelProvider, + modelName: original.modelName, + promptVersion: original.promptVersion, + outputSchemaVersion: original.outputSchemaVersion, + inputSchemaVersion: original.inputSchemaVersion, + maxAttempts: original.maxAttempts, + timeoutMs: original.timeoutMs, + lifecycleStatus: 'queued', + status: 'pending', + queuedAt: new Date(), + attemptCount: 0, + }, + }); + + // 复制 Snapshot + if (original.snapshot) { + await this.prisma.aiJobSnapshot.create({ + data: { + jobId: newJob.id, + jobType: newJob.jobType, + schemaVersion: original.snapshot.schemaVersion, + redactionVersion: original.snapshot.redactionVersion, + content: original.snapshot.content as any, + contentHash: original.snapshot.contentHash, + }, + }); + } + + // 创建 OutboxEvent + await this.prisma.outboxEvent.create({ + data: { + eventType: 'ai.job.enqueue', + aggregateType: 'AiJob', + aggregateId: newJob.id, + dedupeKey: `ai.job.enqueue:${newJob.id}`, + payload: { jobId: newJob.id } as any, + status: 'pending', + availableAt: new Date(), + }, + }); + + return this.toAdminResponse(newJob); + } + + // ── 可观测性 ── + + async getOutboxStatus() { + const counts = await Promise.all([ + this.prisma.outboxEvent.count({ where: { status: 'pending' } }), + this.prisma.outboxEvent.count({ where: { status: 'processing' } }), + this.prisma.outboxEvent.count({ where: { status: 'published' } }), + this.prisma.outboxEvent.count({ where: { status: 'failed' } }), + ]); + return { pending: counts[0], processing: counts[1], published: counts[2], failed: counts[3] }; + } + + // ── 响应构造 ── + + private toPublicResponse(job: any) { + return { + id: job.id, + jobType: job.jobType, + lifecycleStatus: job.lifecycleStatus, + queueName: job.queueName, + priority: job.priority, + progress: job.progress, + createdAt: job.createdAt, + queuedAt: job.queuedAt, + startedAt: job.startedAt, + finishedAt: job.finishedAt, + errorCode: job.errorCode, + publicErrorMessage: job.publicErrorMessage, + artifacts: (job.artifacts || []).map((a: any) => ({ + artifactType: a.artifactType, + artifactId: a.artifactId, + ordinal: a.ordinal, + })), + // 禁止返回：validatedOutput, internalErrorMessage, snapshot, credential, outbox + }; + } + + private toAdminResponse(job: any) { + return { + id: job.id, + userId: job.userId, + jobType: job.jobType, + lifecycleStatus: job.lifecycleStatus, + triggerType: job.triggerType, + queueName: job.queueName, + priority: job.priority, + progress: job.progress, + targetType: job.targetType, + targetId: job.targetId, + parentJobId: job.parentJobId, + attemptCount: job.attemptCount, + maxAttempts: job.maxAttempts, + timeoutMs: job.timeoutMs, + credentialMode: job.credentialMode, + modelProvider: job.modelProvider, + modelName: job.modelName, + createdAt: job.createdAt, + queuedAt: job.queuedAt, + startedAt: job.startedAt, + finishedAt: job.finishedAt, + errorCode: job.errorCode, + publicErrorMessage: job.publicErrorMessage, + internalErrorMessage: job.internalErrorMessage, + validatedOutput: job.validatedOutput, + outputHash: job.outputHash, + artifacts: (job.artifacts || []).map((a: any) => ({ + artifactType: a.artifactType, + artifactId: a.artifactId, + ordinal: a.ordinal, + })), + }; + } +} diff --git a/src/workers/worker-heartbeat.service.spec.ts b/src/workers/worker-heartbeat.service.spec.ts index fc5b6f1..e675512 100644 --- a/src/workers/worker-heartbeat.service.spec.ts +++ b/src/workers/worker-heartbeat.service.spec.ts @@ -3,6 +3,7 @@ import { RedisService } from '../infrastructure/redis/redis.service'; describe('WorkerHeartbeatService', () => { let service: WorkerHeartbeatService; + let mockQ: any; let mockRedis: jest.Mocked>; beforeEach(() => { @@ -15,7 +16,8 @@ describe('WorkerHeartbeatService', () => { }; process.env.WORKER_INSTANCE_ID = 'test-worker-1'; - service = new WorkerHeartbeatService(mockRedis as any); + mockQ = { getActiveCount: jest.fn().mockResolvedValue(0) }; + service = new WorkerHeartbeatService(mockRedis as any, mockQ as any, mockQ as any); }); afterEach(async () => { @@ -61,7 +63,7 @@ describe('WorkerHeartbeatService', () => { it('uses instanceId from WORKER_INSTANCE_ID env var', async () => { process.env.WORKER_INSTANCE_ID = 'custom-instance'; - const svc = new WorkerHeartbeatService(mockRedis as any); + const svc = new WorkerHeartbeatService(mockRedis as any, mockQ as any, mockQ as any); await svc.onModuleInit(); expect(mockRedis.set).toHaveBeenCalledWith( 'worker:heartbeat:custom-instance', diff --git a/src/workers/worker-heartbeat.service.ts b/src/workers/worker-heartbeat.service.ts index 9e281d1..6a3dc85 100644 --- a/src/workers/worker-heartbeat.service.ts +++ b/src/workers/worker-heartbeat.service.ts @@ -1,6 +1,9 @@ import { Injectable, OnModuleInit, OnModuleDestroy, Logger } from '@nestjs/common'; import { RedisService } from '../infrastructure/redis/redis.service'; import { QUEUE_DEFINITIONS } from '../infrastructure/queue/queue-definitions'; +import { QUEUE_AI_INTERACTIVE, QUEUE_AI_BACKGROUND } from '../infrastructure/queue/queue.service'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; import * as os from 'os'; export interface WorkerHeartbeatData { @@ -32,7 +35,11 @@ export class WorkerHeartbeatService implements OnModuleInit, OnModuleDestroy { private static readonly HEARTBEAT_INTERVAL_MS = 30_000; private static readonly HEARTBEAT_TTL_SEC = 90; - constructor(private readonly redis: RedisService) {} + constructor( + private readonly redis: RedisService, + @InjectQueue(QUEUE_AI_INTERACTIVE) private readonly aiInteractiveQ: Queue, + @InjectQueue(QUEUE_AI_BACKGROUND) private readonly aiBackgroundQ: Queue, + ) {} async onModuleInit() { this.startedAt = new Date(); @@ -115,7 +122,7 @@ export class WorkerHeartbeatService implements OnModuleInit, OnModuleDestroy { return WorkerHeartbeatService.heartbeatKeyFor(this.resolveInstanceId()); } - private async beat() { + private async beat(): Promise { const instanceId = this.resolveInstanceId(); const data: WorkerHeartbeatData = { instanceId, @@ -124,7 +131,7 @@ export class WorkerHeartbeatService implements OnModuleInit, OnModuleDestroy { startedAt: this.startedAt.toISOString(), queues: Object.keys(QUEUE_DEFINITIONS), concurrency: Object.values(QUEUE_DEFINITIONS)[0]?.workerOptions.concurrency ?? 1, - activeJobCount: 0, + activeJobCount: (await this.aiInteractiveQ.getActiveCount()) + (await this.aiBackgroundQ.getActiveCount()), hostname: os.hostname(), };