docs: M-AI-03 审计现有 Job 创建、Worker 执行与 Outbox 接入边界
产出 docs/architecture/m-ai-03-current-execution-audit.md。 修正： - 移除死代码 publishAsync() 的 Producer 身份（零调用方） - 移除 NotificationsService.send() 的 Producer 身份（仅写 DB，不入队） - 移除 AdminEventsController 的 Producer 身份（只读+重试，不创建 Job） - 新增孤儿队列节：notification/domain-events/audit-logs 的 Consumer 存在但无活跃 Producer Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wangdl
parent
36903e797e
commit
ed783967e8
474
docs/architecture/m-ai-03-current-execution-audit.md
Normal file
474
docs/architecture/m-ai-03-current-execution-audit.md
Normal file
@ -0,0 +1,474 @@
|
|||||||
|
# M-AI-03 现有执行边界审计
|
||||||
|
|
||||||
|
> 审计日期：2026-06-20
|
||||||
|
> 审计范围：`api-server` 仓库所有 Job 创建、Worker 执行、Queue Producer、Provider 调用、EventBus 与 Outbox 路径
|
||||||
|
> 审计方法：全文搜索 + 逐文件阅读 + 调用链追踪
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 1. 总体架构概览
|
||||||
|
|
||||||
|
```
|
||||||
|
┌────────────────────────────────────────────────────────────┐
|
||||||
|
│ api-server │
|
||||||
|
│ │
|
||||||
|
│ Job System A: Legacy AiJob (BullMQ) │
|
||||||
|
│ ┌──────────────────────────────────────────────────┐ │
|
||||||
|
│ │ AiAnalysisService → AiJob DB → BullMQ │ │
|
||||||
|
│ │ → ai-analysis queue → AiAnalysisWorker │ │
|
||||||
|
│ │ → AiGatewayService → DeepSeek → AiAnalysisResult│ │
|
||||||
|
│ └──────────────────────────────────────────────────┘ │
|
||||||
|
│ │
|
||||||
|
│ Job System B: AiRuntimeJob (REST Poll) │
|
||||||
|
│ ┌──────────────────────────────────────────────────┐ │
|
||||||
|
│ │ UserAiService → AiRuntimeJob DB → Runtime polls │ │
|
||||||
|
│ │ via REST → RuntimeInternalService │ │
|
||||||
|
│ │ → zhixi-heavy-runtime → submitResult │ │
|
||||||
|
│ └──────────────────────────────────────────────────┘ │
|
||||||
|
│ │
|
||||||
|
│ Job System C: DocumentImport (BullMQ) │
|
||||||
|
│ ┌──────────────────────────────────────────────────┐ │
|
||||||
|
│ │ DocumentImportService → DocumentImport DB │ │
|
||||||
|
│ │ → BullMQ document-import queue │ │
|
||||||
|
│ │ → DocumentImportWorker → KnowledgeImportWorkflow│ │
|
||||||
|
│ └──────────────────────────────────────────────────┘ │
|
||||||
|
│ │
|
||||||
|
│ Supporting Queues (BullMQ) │
|
||||||
|
│ ┌──────────────────────────────────────────────────┐ │
|
||||||
|
│ │ notification, domain-events, audit-logs, │ │
|
||||||
|
│ │ file-cleanup │ │
|
||||||
|
│ └──────────────────────────────────────────────────┘ │
|
||||||
|
│ │
|
||||||
|
│ Outbox (exists, unused) │
|
||||||
|
│ ┌──────────────────────────────────────────────────┐ │
|
||||||
|
│ │ OutboxRepository — 表就绪，无生产者，无 Dispatcher │ │
|
||||||
|
│ └──────────────────────────────────────────────────┘ │
|
||||||
|
└────────────────────────────────────────────────────────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
**关键发现**：存在 **两套独立的 Job 系统**（AiJob + AiRuntimeJob），使用完全不同的调度机制（BullMQ vs REST Poll）。DocumentImport 使用独立的 model 和队列。Outbox Repository 已实现但没有任何调用方。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 2. Job Producer 矩阵（仅含活跃生产者）
|
||||||
|
|
||||||
|
| # | Producer Class | Method | Trigger | DB Table | Queue | Payload | 状态写入 | 代码位置 |
|
||||||
|
|---|---------------|--------|---------|----------|-------|---------|----------|----------|
|
||||||
|
| P1 | `AiAnalysisService` | `analyze()` | User API | `AiJob` | `ai-analysis` | `{jobId, userId, type:'active-recall', questionText, knowledgeItemContent, userAnswer}` | `status:pending`, `lifecycleStatus:queued` | `ai-analysis.service.ts:19-31` |
|
||||||
|
| P2 | `AiAnalysisService` | `evaluateFeynman()` | User API | `AiJob` | `ai-analysis` | `{jobId, userId, type:'feynman-evaluation', knowledgeItemTitle, knowledgeItemContent, userExplanation}` | `status:pending`, `lifecycleStatus:queued` | `ai-analysis.service.ts:40-51` |
|
||||||
|
| P3 | `DocumentImportService` | `createImport()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, rawText, fileName}` | `status:QUEUED`(via Repository) | `document-import.service.ts:43-49` |
|
||||||
|
| P4 | `KnowledgeSourceService` | `addSource()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, sourceId, fileName}` | `status:QUEUED`(via Repository) | `knowledge-source.service.ts:43-50` |
|
||||||
|
| P5 | `KnowledgeSourceService` | `triggerParse()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, sourceId, fileName}` | `status:QUEUED`(via Repository) | `knowledge-source.service.ts:90-97` |
|
||||||
|
| P6 | `UserAiService` | `requestJob()` | User API | `AiRuntimeJob` | **无 BullMQ** | REST Poll，不适用 | `status:pending` | `user-ai.service.ts:282-299` |
|
||||||
|
| P7 | `AdminFilesController` | COS cleanup | Admin API | 无 | `file-cleanup` | `{objectKey, bucket, region}` | 无（仅入队） | `admin-files.controller.ts:41` |
|
||||||
|
|
||||||
|
### 死代码（定义为入队方法但零调用方）
|
||||||
|
|
||||||
|
| 方法 | 文件 | 入队目标 | 说明 |
|
||||||
|
|------|------|---------|------|
|
||||||
|
| `EventBusService.publishAsync()` | `event-bus.service.ts:26-35` | `domain-events` | **零调用方** — 全文搜索确认无任何代码调用此方法；`domain-events` 队列当前无活跃生产者 |
|
||||||
|
|
||||||
|
### 孤儿队列（Consumer 已注册但生产者未找到）
|
||||||
|
|
||||||
|
| 队列 | Consumer | 状态 |
|
||||||
|
|------|----------|------|
|
||||||
|
| `notification` | `NotificationWorker` (`workers/notification.worker.ts:8`) | ⚠️ 无 `queue.add('notification', ...)` 调用方；`NotificationsService.send()` 仅写 DB + sync eventBus.publish()，不入队 |
|
||||||
|
| `domain-events` | 无专用 Consumer（`EventBusService.publishAsync` 是唯一入队路径但为零调用方） | ⚠️ 完全闲置 |
|
||||||
|
| `audit-logs` | `AuditLogProcessor` (`modules/admin-audit-log/audit-log.processor.ts:7`) | ⚠️ 无 `queue.add('audit-logs', ...)` 调用方；审计日志走 `prisma.adminAuditLog.create()` 直接写 DB |
|
||||||
|
|
||||||
|
> **注意**：`AdminEventsController` 仅**读取**队列状态（`getWaitingCount`/`getActiveCount`/`getFailed` 等）和**重试**已有失败 Job（`job.retry()`），**从不创建新 Job**。因此不列为 Producer。
|
||||||
|
|
||||||
|
### Producer 代码证据
|
||||||
|
|
||||||
|
**P1–P2** `src/modules/ai-analysis/ai-analysis.service.ts:19-31, 40-51`
|
||||||
|
```typescript
|
||||||
|
const job = await this.repository.createJob(userId, 'active-recall', input.sessionId, input.answerId);
|
||||||
|
await this.queue.add('ai-analysis', { jobId: job.id, userId, type: 'active-recall', ... });
|
||||||
|
```
|
||||||
|
|
||||||
|
**P3** `src/modules/document-import/document-import.service.ts:43-49`
|
||||||
|
```typescript
|
||||||
|
const job = await this.repository.create(dto);
|
||||||
|
await this.queue.add('document-import', { importId: job.id, userId, knowledgeBaseId, rawText, fileName });
|
||||||
|
```
|
||||||
|
|
||||||
|
**P4–P5** `src/modules/knowledge-source/knowledge-source.service.ts:44-50, 91-97`
|
||||||
|
```typescript
|
||||||
|
const importJob = await this.importRepo.create({ ... });
|
||||||
|
await this.queue.add('document-import', { importId: importJob.id, userId, ... });
|
||||||
|
```
|
||||||
|
|
||||||
|
**P6** `src/modules/ai-runtime/user-ai.service.ts:282-299`
|
||||||
|
```typescript
|
||||||
|
const job = await this.prisma.aiRuntimeJob.create({ data: { userId, jobType, status: 'pending', ... } });
|
||||||
|
```
|
||||||
|
|
||||||
|
**P7** `src/modules/files/admin-files.controller.ts:41`
|
||||||
|
```typescript
|
||||||
|
await this.queue.add(QUEUE_FILE_CLEANUP, { objectKey: file.objectKey, bucket: file.bucket, region: 'ap-beijing' });
|
||||||
|
```
|
||||||
|
|
||||||
|
### 死代码与孤儿队列证据
|
||||||
|
|
||||||
|
**`publishAsync()` — 零调用方** `src/common/event-bus/event-bus.service.ts:26-35`
|
||||||
|
```typescript
|
||||||
|
async publishAsync(event: BaseDomainEvent): Promise<string> {
|
||||||
|
if (!this.queue) return '';
|
||||||
|
const job = await this.queue.add('domain-events', { eventType, eventId, payload, occurredAt });
|
||||||
|
return job.id || '';
|
||||||
|
}
|
||||||
|
// 全文搜索确认：整个代码库中无任何代码调用 publishAsync()
|
||||||
|
```
|
||||||
|
|
||||||
|
**`NotificationsService.send()` — 仅写 DB，不入队** `src/modules/notifications/notifications.service.ts:44-49`
|
||||||
|
```typescript
|
||||||
|
async send(data: { userId: string; type: string; title: string; body: string }) {
|
||||||
|
const notification = await this.repository.create(data); // 仅写 DB
|
||||||
|
this.eventBus?.publish(new NotificationSentEvent(...)); // sync 事件，不入队
|
||||||
|
return notification;
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**`AdminEventsController` — 只读 + 重试，不创建新 Job** `src/modules/admin-events/admin-events.controller.ts:1-163`
|
||||||
|
- 所有方法：`getWaitingCount`, `getActiveCount`, `getFailedCount`, `getJob()`, `job.retry()` — 无 `queue.add()`
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 3. Worker / Processor 矩阵
|
||||||
|
|
||||||
|
| # | Worker Class | Queue | 所在 Module | 注入的 Provider | 结果写入 | 发布事件 |
|
||||||
|
|---|-------------|-------|------------|-----------------|---------|---------|
|
||||||
|
| W1 | `AiAnalysisWorker` | `ai-analysis` | `WorkerModule` | `ActiveRecallAnalysisWorkflow`, `FeynmanEvaluationWorkflow`, `AiAnalysisRepository`, `EventBusService?`, `FocusItemsService?` | `AiAnalysisResult` | `ai.analysis.completed` |
|
||||||
|
| W2 | `DocumentImportWorker` | `document-import` | `WorkerModule` | `DocumentImportRepository`, `KnowledgeItemsRepository`, `KnowledgeImportWorkflow`, `RedisService` | `KnowledgeItem` (多个) | 无 |
|
||||||
|
| W3 | `NotificationWorker` | `notification` | `WorkerModule` | `NotificationsService` | `Notification` 记录 | 无 |
|
||||||
|
| W4 | `AuditLogProcessor` | `audit-logs` | `WorkerModule` | `PrismaService` | `AdminAuditLog` | 无 |
|
||||||
|
| W5 | `FileCleanupProcessor` | `file-cleanup` | `WorkerModule` | `CosStorageProvider` | COS 删除 | 无 |
|
||||||
|
|
||||||
|
### Worker 代码证据
|
||||||
|
|
||||||
|
**W1** `src/workers/ai-analysis.worker.ts:18-106`
|
||||||
|
- 行 48: `await this.repository.updateJobStatus(jobId, 'processing')`
|
||||||
|
- 行 59–64: 调用 `feynmanWorkflow.execute()` / `recallWorkflow.execute()`
|
||||||
|
- 行 67–68: `createResult()` + `updateJobStatus(jobId, 'completed')`
|
||||||
|
- 行 72: `this.eventBus?.publish(new AIAnalysisCompleted({...}))`
|
||||||
|
- 行 86–95: 对每个 weakness 创建 `FocusItem`
|
||||||
|
|
||||||
|
**W2** `src/workers/document-import.worker.ts:11-92`
|
||||||
|
- 行 42–45: rawText 为空时直接标记 completed
|
||||||
|
- 行 50–53: 否则 → Redis 写进度，调用 `workflow.execute()`
|
||||||
|
- 行 65–77: 逐个创建 `KnowledgeItem`
|
||||||
|
- 行 79–82: `updateStatus(importId, 'completed')`
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 4. AiRuntimeJob 完整链路（System B — REST Poll）
|
||||||
|
|
||||||
|
```
|
||||||
|
User API → UserAiService.requestJob()
|
||||||
|
→ SnapshotBuilder.buildSnapshot()
|
||||||
|
→ prisma.aiRuntimeJob.create({ status: 'pending' })
|
||||||
|
→ prisma.questionGenerationPlan / flashcardGenerationPlan (if applicable)
|
||||||
|
→ return { jobId, status: 'pending' }
|
||||||
|
|
||||||
|
zhixi-heavy-runtime (external) → RuntimeInternalService.pollJobs()
|
||||||
|
→ prisma.aiRuntimeJob.findMany({ status: 'pending', jobType: { in: [...] } })
|
||||||
|
→ filter by snapshotVersion / outputSchemaVersion capacity
|
||||||
|
→ return { jobs: [...] }
|
||||||
|
|
||||||
|
zhixi-heavy-runtime → RuntimeInternalService.lockJob()
|
||||||
|
→ CAS updateMany(status:pending → status:locked, lockUntil:now+60s)
|
||||||
|
|
||||||
|
zhixi-heavy-runtime → RuntimeInternalService.heartbeatJob()
|
||||||
|
→ updateMany(status:locked → status:running, startedAt:now)
|
||||||
|
→ extend lockUntil (status:running)
|
||||||
|
→ check cancelRequestedAt
|
||||||
|
|
||||||
|
zhixi-heavy-runtime → RuntimeInternalService.getSnapshot()
|
||||||
|
zhixi-heavy-runtime → RuntimeInternalService.resolveCredential()
|
||||||
|
|
||||||
|
zhixi-heavy-runtime → RuntimeInternalService.submitResult()
|
||||||
|
→ prisma.aiRuntimeResult.create()
|
||||||
|
→ prisma.aiRuntimeJob.update(status:succeeded)
|
||||||
|
→ persistResult() → AiLearningAnalysis / WeakPointCandidate / NextActionRecommendation / Quiz / Flashcard
|
||||||
|
→ notifyJobComplete() → Notification
|
||||||
|
|
||||||
|
zhixi-heavy-runtime → RuntimeInternalService.submitFailure()
|
||||||
|
→ retry: status:pending, retryCount++
|
||||||
|
→ exhausted: status:failed, notifyJobComplete()
|
||||||
|
```
|
||||||
|
|
||||||
|
### 关键代码位置
|
||||||
|
|
||||||
|
| 步骤 | 文件 | 行号 |
|
||||||
|
|------|------|------|
|
||||||
|
| 创建 Job | `src/modules/ai-runtime/user-ai.service.ts` | 270–298 |
|
||||||
|
| Poll | `src/modules/ai-runtime/internal/runtime-internal.service.ts` | 22–81 |
|
||||||
|
| Lock (CAS) | 同上 | 85–114 |
|
||||||
|
| Heartbeat | 同上 | 118–149 |
|
||||||
|
| Get Snapshot | 同上 | 153–198 |
|
||||||
|
| Resolve Credential | 同上 | 201–217 |
|
||||||
|
| Submit Result | 同上 | 220–284 |
|
||||||
|
| Persist (by jobType) | 同上 | 286–400 |
|
||||||
|
| Submit Failure | 同上 | 572–625 |
|
||||||
|
| Notify Complete | 同上 | 629–646 |
|
||||||
|
| Invocation Logs | 同上 | 650–694 |
|
||||||
|
| Cancel (user API) | `src/modules/ai-runtime/user-ai.service.ts` | 343–364 |
|
||||||
|
| Reaper (stuck jobs) | `src/modules/ai-runtime/job-reaper.service.ts` | 24–115 |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 5. AiGatewayService — AI Provider 统一网关
|
||||||
|
|
||||||
|
### 调用关系
|
||||||
|
|
||||||
|
```
|
||||||
|
AiGatewayService
|
||||||
|
├── RAG Chat (rag-chat.service.ts:174,249)
|
||||||
|
├── Vector Service (vector.service.ts:147)
|
||||||
|
├── Active Recall Workflow (active-recall-analysis.workflow.ts:15)
|
||||||
|
├── Feynman Workflow (feynman-evaluation.workflow.ts:15)
|
||||||
|
├── Knowledge Import Workflow (knowledge-import.workflow.ts:14)
|
||||||
|
├── Learning Trend Workflow (learning-trend.workflow.ts:28)
|
||||||
|
└── Review Card Workflow (review-card-generation.workflow.ts:15)
|
||||||
|
```
|
||||||
|
|
||||||
|
### 内部结构
|
||||||
|
|
||||||
|
`src/modules/ai/gateway/ai-gateway.service.ts:25-271`
|
||||||
|
|
||||||
|
- **Retry**: `ModelRouter.resolve(tier)` → `preferred` provider → fallback on error → `fallback` provider
|
||||||
|
- **Safety**: `ContentSafetyService.check()` before returning output
|
||||||
|
- **Cost**: `AiCostCalculatorService.calculate()` per call
|
||||||
|
- **Usage Log**: `AiUsageLogService.log()` every attempt
|
||||||
|
- **Event**: `AIUsageRecorded` event on success, `ModelFallbackTriggered` on fallback
|
||||||
|
- **Parse**: 3-layer JSON extraction (direct → markdown fence → regex)
|
||||||
|
- **Stream**: `generateStream()` method for SSE use cases
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 6. 队列配置矩阵
|
||||||
|
|
||||||
|
`src/infrastructure/queue/queue-definitions.ts:94-101` + `src/infrastructure/queue/queue.constants.ts:1-7`
|
||||||
|
|
||||||
|
| Queue Name | concurrency | lockDuration | stalledInterval | maxStalledCount | attempts | backoff | 可环境变量覆盖 |
|
||||||
|
|------------|-------------|-------------|-----------------|-----------------|----------|---------|----------------|
|
||||||
|
| `ai-analysis` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_AI_ANALYSIS_*` |
|
||||||
|
| `document-import` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_DOCUMENT_IMPORT_*` |
|
||||||
|
| `notification` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_NOTIFICATION_*` |
|
||||||
|
| `domain-events` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_DOMAIN_EVENTS_*` |
|
||||||
|
| `audit-logs` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_AUDIT_LOGS_*` |
|
||||||
|
| `file-cleanup` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_FILE_CLEANUP_*` |
|
||||||
|
|
||||||
|
**所有队列 concurrency 均为 1**，没有业务超时（只有 BullMQ 锁机制）。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 7. EventBus 使用矩阵
|
||||||
|
|
||||||
|
`src/common/event-bus/event-bus.service.ts:10-37`
|
||||||
|
|
||||||
|
| 调用方 | Event Type | 发布方式 | 触发时机 |
|
||||||
|
|--------|-----------|---------|---------|
|
||||||
|
| `AiAnalysisWorker` | `ai.analysis.completed` | sync | Job 完成后 |
|
||||||
|
| `AiGatewayService` | `ai.usage.recorded` | sync | AI 调用成功后 |
|
||||||
|
| `AiGatewayService` | `ai.fallback.triggered` | sync | Provider 降级时 |
|
||||||
|
| `GrowthService` | `StreakUpdatedEvent` | sync | 学习连续天数更新 |
|
||||||
|
| `WorkspaceService` | `ItemFavoritedEvent` | sync | 收藏操作 |
|
||||||
|
| `WorkspaceService` | `ItemUnfavoritedEvent` | sync | 取消收藏 |
|
||||||
|
| `WorkspaceService` | `TagCreatedEvent` | sync | 创建标签 |
|
||||||
|
| `WorkspaceService` | `TagDeletedEvent` | sync | 删除标签 |
|
||||||
|
| `WorkspaceService` | `SearchPerformedEvent` | sync | 执行搜索 |
|
||||||
|
| `NotificationsService` | `NotificationReadEvent` | sync | 通知已读 |
|
||||||
|
| `NotificationsService` | `NotificationSentEvent` | sync | 发送通知 |
|
||||||
|
| `NotificationsService` | `NotificationPreferenceChangedEvent` | sync | 偏好变更 |
|
||||||
|
| `QueueService` | `task.enqueued` | sync | 入队后 |
|
||||||
|
|
||||||
|
### 关键发现
|
||||||
|
|
||||||
|
- **全部 sync 发布** — `publishAsync()` 方法已定义但**全文搜索确认为死代码**（零调用方）；`domain-events` 队列当前无活跃生产者
|
||||||
|
- EventBus 当前仅作为 In-Process Event Emitter 使用；其异步域事件能力（BullMQ `domain-events` 队列）处于闲置状态
|
||||||
|
- `QueueService` 中同步发布的 `task.enqueued` 事件无 Subscriber 消费
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 8. Outbox 现状评估
|
||||||
|
|
||||||
|
### Repository 分析
|
||||||
|
|
||||||
|
`src/infrastructure/outbox/outbox.repository.ts:27-139`
|
||||||
|
|
||||||
|
| 能力 | 实现状态 | 备注 |
|
||||||
|
|------|---------|------|
|
||||||
|
| `createInTransaction(tx, input)` | ✅ 已实现 | 接受外部 `Prisma.TransactionClient` |
|
||||||
|
| `findDispatchable(limit)` | ✅ 已实现 | 简单 `findMany`，无锁 |
|
||||||
|
| `markProcessing(eventId, lockedBy)` | ✅ 已实现 | CAS via `updateMany(status:pending → processing)` |
|
||||||
|
| `markPublished(eventId)` | ✅ 已实现 | 简单 update |
|
||||||
|
| `markFailed(eventId, ...)` | ✅ 已实现 | increment attemptCount |
|
||||||
|
| `releaseExpiredLocks(thresholdMs)` | ✅ 已实现 | 重置超时 processing → pending |
|
||||||
|
| **是否有生产写入** | ❌ 无 | **所有搜索返回零调用方** |
|
||||||
|
| **是否有 Dispatcher** | ❌ 无 | 没有 Dispatcher Service/Worker |
|
||||||
|
| **是否支持并发领取** | ⚠️ 部分 | CAS 保证了唯一领取，但 `findDispatchable` 可能多 Worker 读到相同事件（CAS 后有竞态窗口） |
|
||||||
|
| **是否使用 SKIP LOCKED** | ❌ 否 | 使用应用层 CAS（updateMany + status check），非数据库层 SKIP LOCKED |
|
||||||
|
| **重复发布风险** | ⚠️ 存在 | `findDispatchable` 读取后 `markProcessing` 若 CAS 失败不重试；若 Dispatcher 崩溃后重启，`releaseExpiredLocks` 可恢复但非实时 |
|
||||||
|
|
||||||
|
### 表结构 (Prisma)
|
||||||
|
|
||||||
|
`prisma/schema.prisma:2323` — 拥有 `id`, `eventType`, `aggregateType`, `aggregateId`, `dedupeKey`, `payload`, `status`, `attemptCount`, `availableAt`, `lockedAt`, `lockedBy`, `publishedAt`, `lastErrorCode`, `lastErrorMessage`
|
||||||
|
|
||||||
|
**dedupeKey 有 UNIQUE 约束** → 幂等保证存在
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 9. Job 状态机现状
|
||||||
|
|
||||||
|
### AiJob（Legacy）
|
||||||
|
|
||||||
|
`src/modules/ai-analysis/ai-analysis.repository.ts:8-16`
|
||||||
|
|
||||||
|
```
|
||||||
|
pending → processing → completed / failed
|
||||||
|
```
|
||||||
|
|
||||||
|
- `lifecycleStatus` (M-AI-02-10 Shadow Write): `pending→queued`, `processing→running`, `completed→succeeded`, `failed→failed`
|
||||||
|
- 状态由 `AiAnalysisWorker` 直接写入
|
||||||
|
- **无锁机制** — 依赖 BullMQ 内置的 stalled job 检测
|
||||||
|
- **无取消支持**
|
||||||
|
- **无 retry/reaper** — 完全依赖 BullMQ 的 attempts/backoff
|
||||||
|
|
||||||
|
### AiRuntimeJob（新）
|
||||||
|
|
||||||
|
`src/modules/ai-runtime/job-reaper.service.ts` + `runtime-internal.service.ts`
|
||||||
|
|
||||||
|
```
|
||||||
|
pending → locked → running → succeeded / failed / cancelled
|
||||||
|
↑ ↓ ↓
|
||||||
|
└── retried ←── expired ←── (timeout)
|
||||||
|
```
|
||||||
|
|
||||||
|
- 通过 `lockedBy` + `lockUntil` 实现分布式锁
|
||||||
|
- `RuntimeInternalService.lockJob()` CAS 抢锁
|
||||||
|
- `RuntimeInternalService.heartbeatJob()` 续约
|
||||||
|
- `JobReaperService.reap()` 每 30s 收割过期锁和超时 running
|
||||||
|
- `cancelRequestedAt` → `cancelledAt` → 取消路径
|
||||||
|
- 重试：`retryCount < maxRetryCount` → 重置为 `pending`；否则 → `failed`
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 10. 超时 / 重试 / 取消矩阵
|
||||||
|
|
||||||
|
| 系统 | 超时机制 | 重试次数 | 重试间隔 | 取消支持 | 文件位置 |
|
||||||
|
|------|---------|---------|---------|---------|---------|
|
||||||
|
| AiJob (BullMQ) | BullMQ lockDuration=30s, stalledInterval=30s, maxStalledCount=1 | 3 | exponential 1s | ❌ | `queue-definitions.ts:52-57` |
|
||||||
|
| AiRuntimeJob | timeoutSeconds=120s, Reaper 30s | maxRetryCount=3 | N/A (reaper) | ✅ cancelRequestedAt → cancelledAt | `job-reaper.service.ts`, `user-ai.service.ts:343-364` |
|
||||||
|
| DocumentImport (BullMQ) | BullMQ lockDuration=30s | 3 | exponential 1s | ❌ | `queue-definitions.ts:96` |
|
||||||
|
| AiGatewayService | DEFAULT_TIMEOUT_MS=30000, AbortController | tierConfig.maxRetries | sequential attempts | ❌ | `ai-gateway.service.ts:27,51-151` |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 11. 依赖边界分析
|
||||||
|
|
||||||
|
### 必须在 WorkerModule 的模块
|
||||||
|
|
||||||
|
| 模块 | 原因 | 证据 |
|
||||||
|
|------|------|------|
|
||||||
|
| `AiAnalysisWorker` | 仅 Worker 侧 BullMQ Processor，不应在 App 中注册 | `worker.module.ts:74` |
|
||||||
|
| `DocumentImportWorker` | 同上 | `worker.module.ts:75` |
|
||||||
|
| `NotificationWorker` | 同上 | `worker.module.ts:76` |
|
||||||
|
| `AuditLogProcessor` | 仅后台审计日志写入 | `worker.module.ts:77` |
|
||||||
|
| `FileCleanupProcessor` | 仅后台文件清理 | `worker.module.ts:78` |
|
||||||
|
|
||||||
|
### 必须在 AppModule + WorkerModule 共用的模块
|
||||||
|
|
||||||
|
| 模块 | 原因 | 证据 |
|
||||||
|
|------|------|------|
|
||||||
|
| `AiAnalysisModule` | API 侧创建 Job → `AiAnalysisService`；Worker 侧消费 → `AiAnalysisWorker` | `app.module.ts:40`, `worker.module.ts:20` |
|
||||||
|
| `DocumentImportModule` | API 侧创建 Import → `DocumentImportService`；Worker 侧消费 → `DocumentImportWorker` | `app.module.ts:37`, `worker.module.ts:21` |
|
||||||
|
| `AiModule` | API 侧 RAG/Vector 调用 AiGateway；Worker 侧 Workflows 调用 AiGateway | `app.module.ts:10`, `worker.module.ts:8` |
|
||||||
|
| `EventBusModule` | 双向：API 侧发布事件，Worker 侧发布/消费事件 | `app.module.ts:9`, `worker.module.ts:24` |
|
||||||
|
| `NotificationsModule` | API 侧 CRUD；Worker 侧 NotificationWorker | `app.module.ts:44`, `worker.module.ts:23` |
|
||||||
|
|
||||||
|
### 仅在 AppModule 的模块（不涉及 Worker）
|
||||||
|
|
||||||
|
| 模块 | 原因 |
|
||||||
|
|------|------|
|
||||||
|
| `AiRuntimeModule` | REST Poll 模式，Runtime 外部消费，无 BullMQ Worker |
|
||||||
|
|
||||||
|
### 循环依赖检查
|
||||||
|
|
||||||
|
| 路径 | 状态 | 说明 |
|
||||||
|
|------|------|------|
|
||||||
|
| `EventBusService` → `QueueService` | ⚠️ `forwardRef` 解决 | `event-bus.service.ts:15` — `@Inject(forwardRef(() => require(...QueueService)))` |
|
||||||
|
| `AiAnalysisModule` → `AiModule` (via Workflow) → AiGateway → `EventBusService` → `QueueService` | ✅ 单向 | 无需 forwardRef |
|
||||||
|
| API ↔ Worker 循环 | ✅ 无 | Worker 仅消费队列，不调用 API 端点 |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 12. 现状总结与 M-AI-03 风险点
|
||||||
|
|
||||||
|
### 必须保持的旧代码
|
||||||
|
|
||||||
|
1. **`AiAnalysisRepository`** — 当前 AI Job 创建的唯一入口（P1、P2 路径）
|
||||||
|
2. **`AiAnalysisWorker`** — 处理 active-recall 和 feynman-evaluation 的已有业务
|
||||||
|
3. **`DocumentImportWorker`** — 已有文档导入链路
|
||||||
|
4. **`RuntimeInternalService`** — AiRuntimeJob 的 poll/lock/submit 链路（zhixi-heavy-runtime 依赖）
|
||||||
|
5. **`JobReaperService`** — AiRuntimeJob 的过期锁收割
|
||||||
|
6. **All 6 BullMQ queues** — 已有业务依赖
|
||||||
|
|
||||||
|
### 可以独立新增的模块
|
||||||
|
|
||||||
|
1. **新 Job Engine** — 在 M-AI-02 Schema Expand 基础上纯代码层实现
|
||||||
|
2. **新 Outbox Dispatcher** — 独立 Service/Worker，不修改已有队列
|
||||||
|
3. **新 Registry** — 独立 Module，不依赖已有 Processor
|
||||||
|
4. **新状态机** — 独立于现有 `AiAnalysisRepository.STATUS_TO_LIFECYCLE`
|
||||||
|
5. **新 Projector** — 独立于现有 `RuntimeInternalService.persistResult()`
|
||||||
|
|
||||||
|
### M-AI-03 需要避免的冲突
|
||||||
|
|
||||||
|
- **不修改** `AiAnalysisWorker` 的业务逻辑
|
||||||
|
- **不修改** `RuntimeInternalService` 的 REST 接口（heavy-runtime 依赖）
|
||||||
|
- **不修改** `AiGatewayService` 的 provider 调用链
|
||||||
|
- **不修改** BullMQ 队列定义（已有队列保持）
|
||||||
|
- **新 Engine 不接管** 已有 `ai-analysis` 和 `document-import` 队列
|
||||||
|
- **新代码仅在** `AiJob` 表（即 `AiAnalysisJob`），不碰 `AiRuntimeJob` 表
|
||||||
|
|
||||||
|
### 关键风险
|
||||||
|
|
||||||
|
| 风险 | 严重度 | 说明 |
|
||||||
|
|------|--------|------|
|
||||||
|
| 两套 Job 状态不同步 | 🔴 高 | `AiJob.status`(legacy enum) 与 `AiJob.lifecycleStatus`(M-AI-02 新字段) 存在映射但不完整 |
|
||||||
|
| Outbox 无 Dispatcher | 🟡 中 | 需新建但独立 |
|
||||||
|
| `forwardRef` EventBus↔Queue | 🟢 低 | 已有解决方案，不扩大 |
|
||||||
|
| DocumentImport 使用独立 model | 🟡 中 | 不在 M-AI-03 范围，但未来统一需注意 |
|
||||||
|
| RuntimeInternal 无事务保证 | 🔴 高 | result + job update 分两步，非原子 |
|
||||||
|
| 孤儿队列（notification/domain-events/audit-logs） | 🟡 中 | Consumer 已注册但无活跃 Producer；若从未有消息入队，`NotificationWorker` 和 `AuditLogProcessor` 实为闲置进程
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 附录 A：完整文件清单
|
||||||
|
|
||||||
|
| 文件 | 角色 |
|
||||||
|
|------|------|
|
||||||
|
| `src/modules/ai-analysis/ai-analysis.repository.ts` | AiJob 持久层 |
|
||||||
|
| `src/modules/ai-analysis/ai-analysis.service.ts` | AiJob Producer |
|
||||||
|
| `src/modules/ai-analysis/ai-analysis.module.ts` | 模块注册 |
|
||||||
|
| `src/workers/ai-analysis.worker.ts` | AiJob Consumer |
|
||||||
|
| `src/modules/document-import/document-import.service.ts` | Document Import Producer |
|
||||||
|
| `src/modules/document-import/document-import.repository.ts` | Document Import 持久层 |
|
||||||
|
| `src/workers/document-import.worker.ts` | Document Import Consumer |
|
||||||
|
| `src/modules/knowledge-source/knowledge-source.service.ts` | Knowledge Source Producer（间接） |
|
||||||
|
| `src/workers/notification.worker.ts` | Notification Consumer |
|
||||||
|
| `src/modules/admin-audit-log/audit-log.processor.ts` | Audit Log Consumer |
|
||||||
|
| `src/modules/files/file-cleanup.processor.ts` | File Cleanup Consumer |
|
||||||
|
| `src/infrastructure/queue/queue.service.ts` | QueueService — 统一入队入口 |
|
||||||
|
| `src/infrastructure/queue/queue-definitions.ts` | 队列定义注册表 |
|
||||||
|
| `src/infrastructure/queue/queue.constants.ts` | 队列名常量 |
|
||||||
|
| `src/infrastructure/queue/queue.module.ts` | 队列模块（BullMQ 注册） |
|
||||||
|
| `src/infrastructure/outbox/outbox.repository.ts` | OutboxRepository（无调用方） |
|
||||||
|
| `src/common/event-bus/event-bus.service.ts` | EventBus — 同步/异步发布 |
|
||||||
|
| `src/modules/ai/gateway/ai-gateway.service.ts` | AI 网关 — 统一 Provider 路由 |
|
||||||
|
| `src/modules/ai-runtime/internal/runtime-internal.service.ts` | Runtime Internal REST API |
|
||||||
|
| `src/modules/ai-runtime/user-ai.service.ts` | AiRuntimeJob Producer + Cancel |
|
||||||
|
| `src/modules/ai-runtime/job-reaper.service.ts` | AiRuntimeJob Reaper（过期锁收割） |
|
||||||
|
| `src/app.module.ts` | API 进程模块组装 |
|
||||||
|
| `src/worker.module.ts` | Worker 进程模块组装 |
|
||||||
|
| `prisma/schema.prisma` | Prisma Schema（AiJob, AiJobSnapshot, AiJobArtifact, AiRuntimeJob, OutboxEvent 等） |
|
||||||
Loading…
x
Reference in New Issue
Block a user