# fwhQueueNew 设计文档

## 概览

`fwhQueueNew` 是使用 [Asynq](https://github.com/hibiken/asynq) 框架重写的 FWH 任务队列，替代原有基于 Redis Stream 手动实现的 `fwhQueue`。

### 重写目标

1. **消除手动状态管理带来的 bug**（如 `.String()` vs `.Hex()` 导致任务状态不一致）
2. **统一任务监控**（所有任务可在 Asynqmon Web UI 中查看）
3. **简化重试逻辑**（利用 Asynq 内置重试机制替代手动 `for` 循环）
4. **提高可维护性**（声明式任务配置，减少状态同步代码）

---

## 目录结构

```
task/queue/fwhQueueNew/
├── fwhTaskQueue.go              # 初始化：Server、Client、Inspector、常量定义
├── fwhTaskBatchDownload.go      # BatchDownload 主任务 handler
├── fwhTaskCallFile.go           # CallFile 子任务 handler（EmsLog + FaultRecord）
├── fwhTaskSiteAnalysis.go       # SiteAnalysis 任务 handler
├── fwhTaskShutdownStatistics.go # ShutdownStatistics 任务 handler
├── fwhTaskSiteInfoUpdate.go     # SiteInfoUpdate 任务 handler
├── fwhEnqueue.go                # 对外暴露的入队函数
└── DESIGN.md                    # 本文档
```

---

## 架构设计

### 队列拓扑

```
┌─────────────────────────────────────────────────────────┐
│                      Asynq Server                        │
│                     Concurrency: 6                        │
│                                                          │
│  ┌──────────────────┐    ┌───────────────────────────┐  │
│  │  fwh:main (2)    │    │  fwh:download (4)         │  │
│  │                  │    │                           │  │
│  │ • BatchDownload  │    │ • CallFile (EmsLog)       │  │
│  │ • SiteAnalysis   │    │ • CallFile (FaultRecord)  │  │
│  │ • ShutdownStats  │    │                           │  │
│  │ • SiteInfoUpdate │    │                           │  │
│  └──────────────────┘    └───────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
```

- **`fwh:main`**：主队列，2 个 worker，处理业务编排类任务
- **`fwh:download`**：下载队列，4 个 worker，处理文件下载类任务（可并发）

### 任务类型

| 任务类型       | 常量                       | 队列     | 重试 | 超时   |
| -------------- | -------------------------- | -------- | ---- | ------ |
| 批量数据下载   | `fwh:BatchDownload`      | main     | 0    | 120min |
| 文件下载子任务 | `fwh:CallFile`           | download | 3    | 35min  |
| 站点分析       | `fwh:SiteAnalysis`       | main     | 0    | 60min  |
| Shutdown统计   | `fwh:ShutdownStatistics` | main     | 0    | 30min  |
| 站点信息更新   | `fwh:SiteInfoUpdate`     | main     | 0    | 60min  |

---

## BatchDownload 子任务协调机制

### 原有方案（问题）

```
旧方案：Redis Set (standby/failed) + Redis Stream per SN
问题：
1. mainTaskId 使用 .String() vs .Hex() 不一致导致 key 不匹配
2. 提前 return 的路径忘记更新 standby → 主任务永远等待
3. 手动 goroutine per SN 的生命周期管理复杂
```

### 新方案

```
新方案：Redis 计数器 + 失败原因列表

入队时：
  SET fwh:callfile:counter:{mainTaskId} = N (子任务总数)

子任务完成时（无论成功/失败）：
  DECR fwh:callfile:counter:{mainTaskId}
  如果失败：LPUSH fwh:callfile:failures:{mainTaskId} "失败原因"

主任务轮询：
  GET fwh:callfile:counter:{mainTaskId}
  当 counter <= 0 时：
    LRANGE fwh:callfile:failures:{mainTaskId} 0 -1 → 收集失败原因
    DEL counter, failures → 清理
```

**优势**：

- 原子操作（DECR/LPUSH），无竞态
- 不依赖子任务的执行路径，无论 handler 怎么 return/panic，Asynq 都会管理任务生命周期
- 结构简单，只有 2 个 Redis key

---

## 重试策略

### CallFile 子任务

```
Asynq 外层重试（MaxRetry=3）：
  - 第1次失败 → 等待 30s 重试
  - 第2次失败 → 等待 2min 重试
  - 第3次失败 → 标记为最终失败，报告失败原因

内部下载重试（maxCallRetry=3）：
  - 针对 HTTP 下载的短暂网络错误
  - 不涉及重新召唤，只重试下载动作
```

### 其他任务

业务错误（数据格式错误、API 返回失败等）不重试，handler 返回 `nil` 避免 Asynq 重试。只有基础设施错误（Redis 断连等）才返回 `error` 触发重试。

---

## 与旧 fwhQueue 的对比

| 维度       | fwhQueue (旧)                               | fwhQueueNew (新)                       |
| ---------- | ------------------------------------------- | -------------------------------------- |
| 底层       | 手动 Redis Stream                           | Asynq（内部用 Redis List + SortedSet） |
| 并发控制   | 单 worker 顺序 + 手动 goroutine per SN      | `Concurrency` + `Queues` 声明式    |
| 重试       | `XAutoClaim` + `XPendingExt` 手动计数   | `MaxRetry(N)` + 指数退避             |
| 任务去重   | 手动 Redis 分布式锁                         | `asynq.Unique(duration)`             |
| 任务状态   | 内存变量`runningID`/`failedTaskList`    | Asynq 自动管理                         |
| 子任务协调 | Redis Set (standby/failed) 手动移动         | Redis 计数器 DECR                      |
| 监控       | 手动`GetTaskInfoList`                     | Asynqmon Web UI                        |
| Bug 风险   | 高（多种状态同步路径）                      | 低（结构化保证）                       |
| 代码量     | ~800 行（fwhTaskQueue.go + callFwhFile.go） | ~500 行（分布在 7 个文件）             |

---

## 迁移指南

### 需要修改的文件

#### 1. `task/queue/taskQueue.go`

```go
package taskQueue

import (
    task_core "fwh-toolbox-server/task/core"
    "fwh-toolbox-server/task/queue/fwhQueueNew"  // 替换 fwhQueue
    "fwh-toolbox-server/task/queue/simCardQueue"
    "github.com/hibiken/asynq"
)

func InitTaskQueue(tc *task_core.TaskCore, redisOpt asynq.RedisClientOpt) {
    if tc == nil {
        panic("InitTaskQueue failed, TaskCore is nil.")
    }

    fwhQueueNew.InitFwhQueueNew(tc, redisOpt)  // 替换 fwhQueue.InitFwhTaskQueue()
    simCardQueue.InitSimCardTaskQueue(tc)
}
```

#### 2. `task/task.go`

```go
package task

import (
    task_core "fwh-toolbox-server/task/core"
    taskQueue "fwh-toolbox-server/task/queue"
    "github.com/hibiken/asynq"
)

func InitTask(redisOpt TaskRedisOpt) {
    redisClientOpt := asynq.RedisClientOpt{
        Addr:     redisOpt.Addr,
        DB:       redisOpt.DB,
        Username: redisOpt.Username,
        Password: redisOpt.Password,
    }

    tc := task_core.NewTaskCore(redisClientOpt)
    taskQueue.InitTaskQueue(tc, redisClientOpt)  // 传入 redisOpt
}
```

#### 3. `services/batchDownloadServices.go`

```go
// 将：
import "fwh-toolbox-server/task/queue/fwhQueue"
fwhQueue.EnqueueBatchDownloadTask(...)

// 替换为：
import "fwh-toolbox-server/task/queue/fwhQueueNew"
fwhQueueNew.EnqueueBatchDownloadTask(...)
```

#### 4. 其他引用 `fwhQueue.EnqueueXxx` 的地方

全局搜索 `fwhQueue.Enqueue` 替换为 `fwhQueueNew.Enqueue`：

- `services/sitesAnalysisServices.go`
- `scheduler/scheduler.go`（如果有定时触发 ShutdownStatistics）
- `controllers/` 中直接调用入队的地方

#### 5. `main.go` 中的 Asynqmon 配置

无需改动，Asynqmon 已经指向同一个 Redis DB，新队列的任务会自动出现在 UI 中。

---

## 注意事项

1. **Redis DB**：新队列使用与 Asynq Server 相同的 Redis DB（当前为 DB 12），计数器 key 也在同一个 DB 中
2. **并行安全**：`fwh:download` 队列的 4 个 worker 可能同时处理不同 SN 的文件下载，这是安全的（每个 SN 独立）
3. **超时保护**：所有任务都设置了 Timeout，防止 handler 卡死
4. **历史保留**：`Retention(7*24*time.Hour)` 保留已完成任务 7 天，方便在 Asynqmon 中查看历史
5. **平滑迁移**：新旧队列可以共存一段时间，逐步将调用切换到新队列