# Deep Scan 队列功能需求

## 背景

### 当前问题

**前端临时解决方案：**
- 使用 p-limit 控制并发数（maxConcurrent = 2）
- 每个文件启动扫描后等待 2 秒
- 勉强限制同时扫描的文件数量

**存在的限制：**
1. **不可靠**：前端无法准确控制后端资源使用
2. **效率低**：固定延迟 2 秒可能不够或浪费时间
3. **状态不同步**：前端和后端对"正在扫描"的理解不一致
4. **无优先级**：所有文件按顺序处理，无法优先处理重要文件
5. **无错误恢复**：如果某个文件处理失败，没有重试机制
6. **无进度跟踪**：无法准确知道队列中每个文件的处理进度

---

## 需求说明

### 核心需求

**1. 后端队列管理**
- 后端维护一个全局的扫描队列
- 限制同时处理的文件数量（可配置，默认 2-4）
- 自动排队等待，先进先出（FIFO）

**2. 队列状态 API**
- `/api/v1/queue/status` - 获取队列状态
  ```json
  {
    "active_count": 2,
    "max_concurrent": 2,
    "pending_count": 3,
    "total_processed": 10,
    "queue": [
      {
        "file_uuid": "xxx",
        "file_name": "video1.mp4",
        "status": "processing",
        "started_at": "2026-10-07T10:00:00Z",
        "progress": 45
      },
      {
        "file_uuid": "yyy",
        "file_name": "video2.mp4",
        "status": "pending",
        "queued_at": "2026-10-07T10:01:00Z",
        "position": 1
      }
    ]
  }
  ```

**3. 队列控制 API**
- `/api/v1/queue/pause` - 暂停队列
- `/api/v1/queue/resume` - 恢复队列
- `/api/v1/queue/clear` - 清空队列
- `/api/v1/queue/prioritize/{file_uuid}` - 提升优先级

**4. 动态并发控制**
- 根据系统资源（CPU、内存、GPU）自动调整并发数
- 或从配置文件/API 设置 `max_concurrent`

**5. 错误处理与重试**
- 自动重试失败的任务（可配置重试次数）
- 记录失败原因
- 提供手动重试 API

---

## API 设计建议

### 1. 提交扫描任务

**请求：**
```http
POST /api/v1/file/{file_uuid}/deep-scan
Content-Type: application/json

{
  "processors": ["face", "asr", "ocr"],
  "priority": "normal",  // low, normal, high
  "retry_on_failure": true,
  "callback_url": "http://..."  // 可选，完成后通知
}
```

**响应：**
```json
{
  "success": true,
  "task_id": "task_xxx",
  "status": "queued",
  "position": 3,
  "estimated_start": "2026-10-07T10:05:00Z"
}
```

---

### 2. 查询队列状态

**请求：**
```http
GET /api/v1/queue/status
```

**响应：**
```json
{
  "active_count": 2,
  "max_concurrent": 2,
  "pending_count": 3,
  "paused": false,
  "queue": [
    {
      "task_id": "task_xxx",
      "file_uuid": "xxx",
      "file_name": "video1.mp4",
      "status": "processing",
      "started_at": "2026-10-07T10:00:00Z",
      "progress": 45,
      "processors": ["face", "asr"]
    },
    {
      "task_id": "task_yyy",
      "file_uuid": "yyy",
      "file_name": "video2.mp4",
      "status": "pending",
      "queued_at": "2026-10-07T10:01:00Z",
      "position": 1,
      "priority": "normal"
    }
  ],
  "statistics": {
    "total_processed": 10,
    "total_failed": 2,
    "average_time": 120.5
  }
}
```

---

### 3. 任务优先级调整

**请求：**
```http
PUT /api/v1/queue/task/{task_id}/priority
Content-Type: application/json

{
  "priority": "high"
}
```

**响应：**
```json
{
  "success": true,
  "position": 1
}
```

---

### 4. 取消任务

**请求：**
```http
DELETE /api/v1/queue/task/{task_id}
```

**响应：**
```json
{
  "success": true,
  "message": "Task cancelled"
}
```

---

## 数据库设计

### queue_tasks 表

```sql
CREATE TABLE queue_tasks (
  task_id TEXT PRIMARY KEY,
  file_uuid TEXT NOT NULL,
  file_name TEXT,
  status TEXT DEFAULT 'pending',  -- pending, processing, completed, failed
  priority TEXT DEFAULT 'normal',
  processors JSONB,
  progress INTEGER DEFAULT 0,
  queued_at TIMESTAMP DEFAULT NOW(),
  started_at TIMESTAMP,
  completed_at TIMESTAMP,
  error_message TEXT,
  retry_count INTEGER DEFAULT 0,
  max_retries INTEGER DEFAULT 3,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_queue_status ON queue_tasks(status);
CREATE INDEX idx_queue_priority ON queue_tasks(priority, queued_at);
```

---

## 实现建议

### 后端架构

```
Queue Manager (Singleton)
├── Task Queue (Priority Queue)
├── Worker Pool (2-4 workers)
├── Resource Monitor
│   ├── CPU Usage
│   ├── Memory Usage
│   └── GPU Usage
└── Scheduler
    ├── Dynamic Concurrency Adjustment
    └── Task Retry Logic
```

### 核心组件

**1. QueueManager**
```rust
pub struct QueueManager {
    tasks: Arc<Mutex<VecDeque<QueueTask>>>,
    active_tasks: Arc<Mutex<HashMap<TaskId, JoinHandle>>>,
    max_concurrent: Arc<AtomicU32>,
    worker_semaphore: Arc<Semaphore>,
}

impl QueueManager {
    pub async fn submit_task(&self, task: QueueTask) -> Result<TaskId>;
    pub async fn get_status(&self) -> QueueStatus;
    pub async fn cancel_task(&self, task_id: TaskId) -> Result<()>;
    pub async fn adjust_concurrency(&self, max: u32);
}
```

**2. ResourceMonitor**
```rust
pub struct ResourceMonitor {
    cpu_threshold: f64,  // 80%
    memory_threshold: f64,  // 80%
    gpu_threshold: f64,  // 90%
}

impl ResourceMonitor {
    pub fn check_resources(&self) -> ResourceStatus;
    pub fn suggest_concurrency(&self) -> u32;
}
```

---

## 前端集成

### 简化后的前端代码

```typescript
async function batchDeepScan() {
  const filesToProcess = selectedRegisteredFiles.value
  
  // 提交所有任务到后端队列
  for (const f of filesToProcess) {
    await apiCall('deep_scan', { 
      fileUuid: f.file_uuid,
      priority: 'normal'
    })
  }
  
  // 轮询队列状态
  const interval = setInterval(async () => {
    const status = await apiCall('get_queue_status')
    actionMsg.value = `Processing: ${status.active_count} active, ${status.pending_count} queuing...`
    
    if (status.pending_count === 0 && status.active_count === 0) {
      clearInterval(interval)
      actionMsg.value = 'All tasks completed'
    }
  }, 2000)
}
```

---

## 预期效果

### 用户体验改进

**当前（前端队列）：**
- ❌ 不准确的并发控制
- ❌ 固定延迟效率低
- ❌ 无法查看队列状态
- ❌ 无法调整优先级
- ❌ 无法恢复失败任务

**改进后（后端队列）：**
- ✅ 准确的并发控制
- ✅ 自动资源管理
- ✅ 实时队列状态查看
- ✅ 灵活的优先级调整
- ✅ 自动重试机制
- ✅ 错误恢复能力

---

## 配置选项

### momentry.toml

```toml
[queue]
# 最大并发处理数
max_concurrent = 2

# 是否启用动态调整
dynamic_concurrency = true

# 资源阈值
cpu_threshold = 80.0
memory_threshold = 80.0
gpu_threshold = 90.0

# 重试配置
max_retries = 3
retry_delay = 10  # seconds

# 任务超时
task_timeout = 3600  # seconds
```

---

## 里程碑

### Phase 1: 基础队列功能
- 队列数据结构和 API
- 基本的并发控制
- 任务状态跟踪

### Phase 2: 高级功能
- 优先级队列
- 资源监控
- 动态并发调整

### Phase 3: 错误处理
- 自动重试
- 错误日志
- 手动干预 API

---

## 参考实现

**类似项目：**
- Celery (Python)
- Bull (Node.js)
- Sidekiq (Ruby)

**关键特性：**
- 持久化队列
- 优先级支持
- 进度跟踪
- 错误处理

---

## 联系方式

**需求提出者：** Momentry Studio Team  
**日期：** 2026-10-07  
**优先级：** High  
**预计工期：** 2-3 周