# 异步任务系统 + 并发拉取批量写入

## Context

当前数据入库通过 CLI `cmd/ingest` 串行执行，全量 1035 只转债约需 1 小时。用户要求：
1. Web 控制页增加全量/增量更新按钮，显示数据库数据时间
2. 后端引入 Redis，异步执行入库任务，任务状态实时写 Redis
3. 前端轮询后端接口实时刷新任务状态
4. 任务和结果同时持久化到 SQLite，Web 可查询任务列表
5. **核心优化**：并发拉取（多 worker 同时从 API 取数据）+ 批量写入（20 只一批统一写 SQLite），替代串行模式

目标：全量入库从 1 小时降至约 5 分钟，且支持 Web 端操作。

## 架构概览

```
前端(轮询2s) → POST /api/tasks → TaskController → JobManager.Start()
                                                      ↓ (异步 goroutine)
                                            Runner: N workers 并发拉取
                                                      ↓ resultCh
                                            Collector: 单 goroutine 批量写 SQLite
                                                      ↓ 每20只
                                            Store.SaveKlinesBatch (单事务)
                                                      ↓
                                            Redis(实时进度) + SQLite(持久化)
```

**关键约束与对策**：
- `sqlite.Store.SetMaxOpenConns(1)` → 单一 collector goroutine 串行写入（worker 不直接写 DB）
- `eastmoney.Client.throttle()` 非线程安全（`lastReq` 无锁）→ 加 sync.Mutex + 可配置间隔，job 用 `NewClientWithThrottle(0)` 关闭内部节流，由全局限速器统一控制
- 预加载优化：`ListCodes()` + `LastKlineDates()`（新增 GROUP BY 一次查询）替代 600 次逐只查询，避免占用唯一连接堵塞写入
- `MarketController` 每请求 `sqlite.Open` → 改为注入共享 `*sqlite.Store`，消除连接池争用

## 实现步骤

### 阶段 A：基础设施

1. **修改 `internal/datasource/eastmoney/client.go`**：`Client` 加 `sync.Mutex` + `interval time.Duration` 字段；`throttle()` 加锁；新增 `NewClientWithThrottle(interval)` 构造函数（`NewClient()` 行为不变，默认 2s）

2. **新增依赖**：`go get github.com/redis/go-redis/v9`，提升 `github.com/google/uuid` 为直接依赖

3. **新建 `internal/cache/cache.go`**：`Task` 结构体（ID/Type/Status/Total/Done/Success/Failed/FailedCodes/Error/时间戳）、`Cache` 接口（GetTask/SetTask/DeleteTask/ListActiveTaskIDs/AddActiveTask/RemoveActiveTask/Ping）、`NoopCache` 空实现（Redis 降级用）

4. **新建 `internal/cache/redis.go`**：`RedisCache` 实现，键设计 `stock-analysis:task:{id}` → Task JSON（TTL 24h）、`stock-analysis:tasks:active` → SET

### 阶段 B：存储层扩展

5. **修改 `internal/storage/sqlite/store.go`**：
   - `schemaSQL` 追加 `ingest_task` 表（id/type/status/total/done/success/failed/failed_codes/params/error/started_at/finished_at）
   - 新增 `SaveKlinesBatch(ctx, batch map[string][]model.Kline)`：单事务批量写入多只转债K线，带 3 次重试
   - 新增 `LastKlineDates(ctx) map[string]string`：`SELECT code, MAX(date) FROM convertible_kline GROUP BY code` 一次查询

6. **新建 `internal/storage/sqlite/task.go`**：任务 CRUD（SaveTask/UpdateTaskProgress/FinishTask/GetTask/ListTasks），`ListTasks` 按 `started_at DESC` 排序

### 阶段 C：任务执行核心

7. **新建 `internal/job/store.go`**：定义 `Store` 接口（含 K线批量写 + 任务 CRUD），编译期断言 `*sqlite.Store` 实现它

8. **新建 `internal/job/ingest.go`**：`resolveCandidates`（full 走 eastmoney datacenter、incremental 走 store.ListCodes）、`yearOf`、`resolveIncrementalBeg`（从 cmd/ingest 提取）

9. **新建 `internal/job/runner.go`**：核心并发逻辑
   - `run()`：预加载 skipSet/lastDateMap → 启动 collector goroutine → N worker 并发拉取 → 等待完成 → flush 残余 batch
   - `collect()`：单 goroutine，从 resultCh 收集，每 20 只调 `SaveKlinesBatch`，调 `reportProgress`
   - `fetchOne()`：单只转债拉取（全量用 FetchFullKline，增量用 FetchIncrementalKline）
   - `reportProgress()`：每次写 Redis，每 5 只写 SQLite（减少写压力）

10. **新建 `internal/job/job.go`**：`Manager` 管理任务注册表 + 全局限速器 `gate`（令牌桶：400ms 间隔，突发 4）
    - `Start(ctx, type, since, limit)`：创建 Task → 存 SQLite+Redis → 异步启动 Runner
    - `Cancel(id)`：调用 context.CancelFunc
    - `Get(ctx, id)`：Redis 优先，回退 SQLite
    - `List(ctx, limit)`：SQLite 历史 + Redis 实时覆盖运行中任务
    - `StopAll()`：服务关闭时优雅停止

### 阶段 D：HTTP 层

11. **新建 `internal/controller/task.go`**：`TaskController`，4 个 handler
    - `POST /api/tasks`：创建任务 `{"type":"full"|"incremental","since":"","limit":0}`
    - `GET /api/tasks`：任务列表
    - `GET /api/tasks/{id}`：任务状态
    - `POST /api/tasks/{id}/cancel`：取消任务

12. **修改 `internal/controller/api.go`**：`MarketController` 字段 `dbPath string` → `store *sqlite.Store`；`NewMarketController(svc, store)`；移除 handler 中临时 `sqlite.Open`

13. **修改 `internal/router/router.go`**：`New` 签名增加 `tc *TaskController`；注册 4 条 task 路由（均 `requireAuth`）

14. **修改 `cmd/server/main.go`**：
    - 共享 `sqlite.Open(dbPath)`（替代 controller 临时 open）
    - `initCache()`：读 `REDIS_ADDR`（默认 localhost:6379），连不上降级 `NoopCache`
    - `initJobManager()`：创建无节流 eastmoney client + fallback + Manager
    - `defer mgr.StopAll()`
    - Docker 启动 Redis：`docker run -d --name stock-redis -p 6379:6379 redis:7-alpine`

### 阶段 E：前端

15. **修改 `web/style.css`**：新增 `.progress-bar`/`.progress-fill`/`.badge-*`（running/success/failed/cancelled）样式

16. **修改 `web/index.html` 数据管理 Tab**：
    - 保留现有状态卡片
    - 新增工具栏：全量更新、增量更新按钮 + since 日期输入（增量可选）
    - 新增任务列表表格：ID（截断）、类型、状态徽章、进度条、完成/总数、时间、取消按钮
    - JS：`createTask(type)` → POST /api/tasks → 启动轮询 `pollTasks()`（2s 间隔，有 running 任务时持续，无则停止）
    - `renderTaskTable(tasks)`：渲染表格 + 进度条
    - `cancelTask(id)` → POST cancel

## 关键文件

| 文件 | 操作 | 说明 |
|---|---|---|
| `internal/datasource/eastmoney/client.go` | 修改 | throttle 加锁 + 可配置间隔 |
| `internal/cache/cache.go` | 新建 | Cache 接口 + Task 模型 + NoopCache |
| `internal/cache/redis.go` | 新建 | Redis 实现 |
| `internal/storage/sqlite/store.go` | 修改 | 新增 ingest_task 表 + SaveKlinesBatch + LastKlineDates |
| `internal/storage/sqlite/task.go` | 新建 | 任务 CRUD |
| `internal/job/store.go` | 新建 | Store 接口 |
| `internal/job/ingest.go` | 新建 | 候选列表解析 + 增量起点 |
| `internal/job/runner.go` | 新建 | worker 池 + collector |
| `internal/job/job.go` | 新建 | Manager + 限速器 |
| `internal/controller/task.go` | 新建 | TaskController |
| `internal/controller/api.go` | 修改 | MarketController 注入共享 Store |
| `internal/router/router.go` | 修改 | 注册 task 路由 |
| `cmd/server/main.go` | 修改 | 装配 Redis/Manager/共享Store |
| `web/index.html` | 修改 | 数据管理 Tab 增强 |
| `web/style.css` | 修改 | 进度条 + 徽章样式 |

## 并发拉取 + 批量写入流程

```
Manager.Start(full)
  → 创建 Task，存 SQLite + Redis
  → go runner.run(ctx, task)

runner.run:
  1. resolveCandidates() → 全量列表或已入库 codes
  2. 预加载：skipSet = ListCodes()，lastDateMap = LastKlineDates()
  3. resultCh = make(chan fetchResult, workers*2)
  4. go collect(resultCh)  // 单一 SQLite 写者
  5. for code in candidates:
       sem <- {}  // worker 槽
       go fetchOne(code):  // 并发拉取
         <-gate  // 全局限速令牌
         klines = ds.FetchKline(code)
         resultCh <- {code, klines, err}
         <-sem
  6. wg.Wait(); close(resultCh)
  7. flush 残余 batch
  8. 更新 Task 状态 → SQLite + Redis

collect:  // 单 goroutine
  batch = {}
  for res in resultCh:
    task.Done++
    if err: task.Failed++; failedCodes.append(code)
    else: task.Success++; batch[code] = klines
    if len(batch) >= 20: store.SaveKlinesBatch(batch); batch = {}
    reportProgress(task)  // Redis 每次 + SQLite 每5只
```

## 验证方法

1. **编译**：`go build ./...` 每阶段验证
2. **竞态检测**：`go test -race ./internal/...`
3. **端到端**：
   - Docker 启动 Redis
   - 启动 server，admin 登录
   - 数据管理 Tab → 点"增量更新" → 任务表出现 running 行 → 进度条实时增长 → 完成
   - 点"全量更新" → 两个任务并行
   - 点"取消" → 状态变 cancelled
   - 重启 server → 任务列表仍在（SQLite 持久化）
   - 停 Redis → 任务仍能跑（NoopCache 降级）
4. **性能**：50 只转债（limit=50）应在 1 分钟内完成