mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-08 08:36:37 +08:00
wavelet init
This commit is contained in:
ryan
@@ -0,0 +1,121 @@
|
||||
---
|
||||
name: "new-async-task"
|
||||
description: "Wavelet 项目专用：新增或修改 Asynq 异步任务、后台任务、定时任务、任务元数据、TaskHandler、TaskParam、PayloadValidator、AppendLog、任务重试、任务执行记录或 Admin 任务 API 时必须使用。"
|
||||
---
|
||||
|
||||
# 异步任务开发
|
||||
|
||||
开始前阅读根目录 `AGENTS.md`。只修改任务相关链路，遵守项目路由、日志、数据库迁移和质量门禁要求。
|
||||
|
||||
## 开始前
|
||||
|
||||
按任务范围检查当前实现：
|
||||
|
||||
- `internal/task/handler.go`：`TaskHandler`、`TaskResult`、`PayloadValidator`
|
||||
- `internal/task/meta.go`：`TaskMeta`、`TaskParam`
|
||||
- `internal/task/executor.go`：下发、执行、日志、重试、`OnTaskCompleted` 订阅
|
||||
- `internal/task/handlers/register.go`：Handler 和元数据注册（由 bootstrap 调用）
|
||||
- `internal/bootstrap/bootstrap.go`：任务注册与进程级装配入口
|
||||
- `internal/task/worker/worker.go`：Worker 路由和队列
|
||||
- `internal/task/scheduler/scheduler.go`：定时调度
|
||||
- `internal/apps/admin/task/routers.go`：Admin 任务 API
|
||||
- `internal/model/task_execution.go`：执行记录和日志持久化
|
||||
|
||||
需要模板时阅读 [references/CODE-EXAMPLES.md](references/CODE-EXAMPLES.md)。
|
||||
|
||||
## 实现要求
|
||||
|
||||
### 任务定义
|
||||
|
||||
- 在 `internal/apps/<module>/tasks.go` 定义任务类型、Admin 任务类型和 `TaskMeta`。
|
||||
- Asynq 任务类型使用 `<module>:<action>` 格式。
|
||||
- 完整设置 `Type`、`AsynqTask`、`Name`、`Description`、`MaxRetry`、`Queue`、`Retryable`。
|
||||
- 有参数任务必须定义 payload struct。
|
||||
- `TaskParam.Name` 必须与 payload JSON tag 一致。
|
||||
- `TaskParam` 只描述前端表单，不代替服务端校验。
|
||||
|
||||
### Handler
|
||||
|
||||
- Handler 必须实现 `task.TaskHandler`。
|
||||
- 有参数任务必须实现 `task.PayloadValidator`，负责校验和标准化 Admin 下发参数。
|
||||
- `Execute` 必须再次解析 payload；不要假设入口一定经过 Admin 校验。
|
||||
- 成功返回 `&task.TaskResult{Message: ..., Detail: ...}`。
|
||||
- 失败返回 error，由任务框架处理状态和重试。
|
||||
- 不要吞掉关键错误。
|
||||
- 复杂 SQL 放到 `internal/model/` 或模块内的业务服务层（如 `internal/apps/<module>/service.go` 或 `logics.go`）。
|
||||
|
||||
### 注册
|
||||
|
||||
- 在 `internal/task/handlers/register.go` 同时注册 Handler 和 `TaskMeta`。
|
||||
- 不要在其他位置单独注册任务。
|
||||
- **禁止**在业务包 `routers.go` 或 `init()` 中调用 `task.RegisterHandler`；统一由 `bootstrap.RegisterTasks()` → `taskhandlers.Register()` 在进程启动时装配。
|
||||
- 任务完成钩子（如 push 通知）通过 `task.OnTaskCompleted` 注册，在 `bootstrap.RegisterTaskListeners()` 中装配（Worker/`all` 进程）。
|
||||
|
||||
### 进程装配分工
|
||||
|
||||
| 进程 | 注册入口 |
|
||||
| :--- | :--- |
|
||||
| `api` | `cmd/api.go` → `bootstrap.RegisterAPI()`（含 `RegisterTasks`） |
|
||||
| `worker` | `worker.StartWorker()` → `bootstrap.RegisterWorker()`（含 `RegisterTasks` + `RegisterTaskListeners`） |
|
||||
| `scheduler` | `scheduler.StartScheduler()` → `bootstrap.RegisterScheduler()` |
|
||||
| `all` | `cmd/all.go` → `bootstrap.RegisterAll()` |
|
||||
|
||||
所有 `Register*` 使用 `sync.Once`，重复调用安全。
|
||||
|
||||
### 测试
|
||||
|
||||
- 依赖已注册任务类型或 Handler 的测试（如 `internal/apps/admin/task/routers_test.go`），必须在 setup 中显式调用 `bootstrap.RegisterTasks()`。
|
||||
- 不得依赖 `init()` 副作用或 import 链触发注册。
|
||||
|
||||
## 日志要求
|
||||
|
||||
- 在 `TaskHandler.Execute` 中使用 `task.AppendLog(ctx, format, args...)`。
|
||||
- 记录任务开始、参数摘要、批次进度、关键状态、可继续错误和完成摘要。
|
||||
- 批量处理按批次记录；禁止为大循环中的每条数据写日志。
|
||||
- 不要直接修改任务日志的 Redis key 或 `w_task_executions.log`。
|
||||
|
||||
日志框架约束：
|
||||
|
||||
- 执行状态实时写入数据库：`pending`、`running`、`succeeded`、`failed`。
|
||||
- 实时日志写入 Redis，每个任务最多保留最近 1000 行。
|
||||
- Redis 日志 TTL 为 24 小时，每次追加时刷新。
|
||||
- 查询时优先返回 Redis 日志，Redis 不存在时读取数据库。
|
||||
- 任务成功或自动重试耗尽后，将日志写入数据库并删除 Redis 缓冲。
|
||||
- 自动重试期间保留同一 taskID 的 Redis 日志。
|
||||
|
||||
## 重试要求
|
||||
|
||||
- Handler 返回 error 以触发 Asynq 自动重试。
|
||||
- 不要在 Handler 内自行实现重复重试循环。
|
||||
- Admin 手动重试只允许：
|
||||
- 原任务状态为 `failed`
|
||||
- `Retryable=true`
|
||||
- `RetryCount < MaxRetry`
|
||||
- 修改重试行为时同时检查：
|
||||
- `internal/task/executor.go`
|
||||
- `internal/model/task_execution.go`
|
||||
- `internal/apps/admin/task/routers.go`
|
||||
- 前端任务执行列表
|
||||
|
||||
## 定时任务
|
||||
|
||||
- 默认定时任务必须通过 Goose SQL 迁移写入 `schedules`。
|
||||
- PostgreSQL 和 SQLite 迁移必须同时提供。
|
||||
- 初始化 SQL 必须幂等。
|
||||
- 涉及迁移时使用 `database-migration` skill。
|
||||
|
||||
## Admin API
|
||||
|
||||
- Handler 放在现有 Admin task 模块或 `internal/apps/admin/<module>/`。
|
||||
- 路由只在 `internal/router/router.go` 注册。
|
||||
- 响应保持 `{ "error_msg": "", "data": ... }`。
|
||||
- 分页数据保持 `{ "total": 0, "results": [] }`。
|
||||
- Swagger 注释必须完整；API 变化后运行 `make swagger`。
|
||||
|
||||
## 前端
|
||||
|
||||
- 仅任务元数据变化时，优先复用现有动态任务表单，不新增页面。
|
||||
- API 调用必须通过 `frontend/lib/services/`。
|
||||
- 修改 shadcn/ui 时使用 `shadcn` skill。
|
||||
- 不使用 `any`。
|
||||
- 页面根容器使用 `w-full`，不添加页面级 `max-w-*`。
|
||||
@@ -0,0 +1,277 @@
|
||||
# Wavelet 异步任务代码示例
|
||||
|
||||
这些示例用于新增或修改 Wavelet Asynq 任务时快速套用。复制前先对照当前代码，因为任务框架可能随项目演进。
|
||||
|
||||
## 任务元数据与常量定义
|
||||
|
||||
在对应的业务包 `internal/apps/<module>/tasks.go` 中定义 Asynq task type、Admin task type 和 `TaskMeta`。
|
||||
|
||||
```go
|
||||
package upload
|
||||
|
||||
import (
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
)
|
||||
|
||||
// 异步任务类型标识。格式建议为 "{module}:{action}"。
|
||||
const CleanupUnusedUploadsTask = "upload:cleanup_unused"
|
||||
|
||||
// 管理员可下发的任务类型标识。用于 Admin API 的 task_type。
|
||||
const TaskTypeCleanupUploads = "cleanup_unused_uploads"
|
||||
|
||||
// CleanupUnusedUploadsMeta 任务元数据
|
||||
var CleanupUnusedUploadsMeta = task.TaskMeta{
|
||||
Type: TaskTypeCleanupUploads,
|
||||
AsynqTask: CleanupUnusedUploadsTask,
|
||||
Name: "清理未使用上传",
|
||||
Description: "清理超过1小时未使用的上传文件",
|
||||
SupportsTime: false,
|
||||
MaxRetry: task.DefaultMaxRetry,
|
||||
Queue: task.QueueDefault,
|
||||
Retryable: true,
|
||||
}
|
||||
```
|
||||
|
||||
带参数任务把前端表单元数据放在 `Params`。`Name` 必须和 payload JSON tag 对齐。
|
||||
|
||||
```go
|
||||
{
|
||||
Type: TaskTypeSendEmail,
|
||||
AsynqTask: SendEmailTask,
|
||||
Name: "发送邮件",
|
||||
Description: "异步发送系统邮件",
|
||||
SupportsTime: false,
|
||||
MaxRetry: defaultMaxRetry,
|
||||
Queue: QueueDefault,
|
||||
Retryable: true,
|
||||
Params: []TaskParam{
|
||||
{
|
||||
Name: "to",
|
||||
Label: "接收邮箱 (To)",
|
||||
Type: "string",
|
||||
Required: true,
|
||||
Placeholder: "receiver@example.com",
|
||||
Description: "接收邮件的目标邮箱地址",
|
||||
},
|
||||
{
|
||||
Name: "subject",
|
||||
Label: "邮件主题 (Subject)",
|
||||
Type: "string",
|
||||
Required: true,
|
||||
Placeholder: "请输入邮件主题",
|
||||
Description: "发送邮件的主题标题",
|
||||
},
|
||||
{
|
||||
Name: "body",
|
||||
Label: "邮件内容 (Body)",
|
||||
Type: "text",
|
||||
Required: true,
|
||||
Placeholder: "请输入邮件内容",
|
||||
Description: "发送邮件的内容主体",
|
||||
},
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
## 无参数 Handler
|
||||
|
||||
放在对应业务模块，例如 `internal/apps/upload/tasks.go`。
|
||||
|
||||
```go
|
||||
package upload
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
)
|
||||
|
||||
type CleanupUnusedUploadsHandler struct{}
|
||||
|
||||
func (h *CleanupUnusedUploadsHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
|
||||
task.AppendLog(ctx, "开始扫描未使用上传")
|
||||
|
||||
// 调用 model/service 完成业务逻辑。
|
||||
// 批量处理时按批次记录日志，不要每条记录都 AppendLog。
|
||||
|
||||
msg := "清理完成"
|
||||
task.AppendLog(ctx, "%s", msg)
|
||||
return &task.TaskResult{Message: msg}, nil
|
||||
}
|
||||
```
|
||||
|
||||
## 带参数 Handler
|
||||
|
||||
实现 `PayloadValidator` 做 Admin 下发时的服务端校验和标准化。`Execute` 仍然解析 payload，因为 Scheduler 和 Retry 不一定经过 Admin 校验路径。
|
||||
|
||||
```go
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
)
|
||||
|
||||
type SendEmailPayload struct {
|
||||
To string `json:"to"`
|
||||
Subject string `json:"subject"`
|
||||
Body string `json:"body"`
|
||||
}
|
||||
|
||||
type SendEmailHandler struct{}
|
||||
|
||||
func (h *SendEmailHandler) ValidatePayload(payload []byte) ([]byte, error) {
|
||||
if len(payload) == 0 {
|
||||
return nil, errors.New("任务参数不能为空")
|
||||
}
|
||||
|
||||
var req SendEmailPayload
|
||||
if err := json.Unmarshal(payload, &req); err != nil {
|
||||
return nil, fmt.Errorf("无效的 JSON 格式: %w", err)
|
||||
}
|
||||
|
||||
req.To = strings.TrimSpace(req.To)
|
||||
req.Subject = strings.TrimSpace(req.Subject)
|
||||
req.Body = strings.TrimSpace(req.Body)
|
||||
if req.To == "" || req.Subject == "" || req.Body == "" {
|
||||
return nil, errors.New("to、subject、body 不能为空")
|
||||
}
|
||||
|
||||
return json.Marshal(req)
|
||||
}
|
||||
|
||||
func (h *SendEmailHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
|
||||
var req SendEmailPayload
|
||||
if err := json.Unmarshal(payload, &req); err != nil {
|
||||
return nil, fmt.Errorf("解析任务参数: %w", err)
|
||||
}
|
||||
|
||||
task.AppendLog(ctx, "开始发送邮件到: %s", req.To)
|
||||
|
||||
// 调用业务服务发送邮件。
|
||||
|
||||
msg := fmt.Sprintf("邮件成功发送至: %s", req.To)
|
||||
task.AppendLog(ctx, "%s", msg)
|
||||
return &task.TaskResult{Message: msg}, nil
|
||||
}
|
||||
```
|
||||
|
||||
## 统一注册
|
||||
|
||||
在 `internal/task/handlers/register.go` 注册。Admin dispatch 的 `ValidateAndNormalizePayload` 和 Worker 执行都依赖这里。
|
||||
|
||||
```go
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload"
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/user"
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
)
|
||||
|
||||
func Register() {
|
||||
task.RegisterHandler(task.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{})
|
||||
task.RegisterHandler(task.SendEmailTask, &user.SendEmailHandler{})
|
||||
}
|
||||
```
|
||||
|
||||
## Cron 调度和配置
|
||||
|
||||
系统默认的定时任务必须通过 Goose SQL 迁移初始化插入到 `schedules` 表。
|
||||
|
||||
在 `internal/db/migrator/goose/postgres` 下的示例：
|
||||
|
||||
```sql
|
||||
-- +goose Up
|
||||
INSERT INTO schedules (id, name, task_type, cron, payload, is_active, created_at, updated_at)
|
||||
VALUES (1, '清理未使用上传', 'cleanup_unused_uploads', '0 */2 * * *', '{}', TRUE, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
ON CONFLICT (id) DO NOTHING;
|
||||
|
||||
-- +goose Down
|
||||
-- 根据业务需求决定是否需要在此删除
|
||||
```
|
||||
|
||||
对于 `sqlite` 也可以使用类似的 `INSERT INTO ... ON CONFLICT(id) DO NOTHING` 语法。数据库更新后，后端会自动热重载调度器。
|
||||
|
||||
## Handler 测试
|
||||
|
||||
带参数任务至少覆盖合法 payload、空 payload、非法 JSON、缺失必填和标准化。
|
||||
|
||||
```go
|
||||
func TestSendEmailHandlerValidatePayload(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
payload []byte
|
||||
want SendEmailPayload
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "valid payload is normalized",
|
||||
payload: []byte(`{"to":" user@example.com ","subject":" hi ","body":" body "}`),
|
||||
want: SendEmailPayload{
|
||||
To: "user@example.com",
|
||||
Subject: "hi",
|
||||
Body: "body",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "empty payload",
|
||||
payload: nil,
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "invalid json",
|
||||
payload: []byte(`{`),
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "missing required field",
|
||||
payload: []byte(`{"to":"user@example.com","subject":"","body":"body"}`),
|
||||
wantErr: true,
|
||||
},
|
||||
}
|
||||
|
||||
h := &SendEmailHandler{}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
gotPayload, err := h.ValidatePayload(tt.payload)
|
||||
if gotErr := err != nil; gotErr != tt.wantErr {
|
||||
t.Fatalf("ValidatePayload(%s) error = %v, want error presence = %t", tt.payload, err, tt.wantErr)
|
||||
}
|
||||
if tt.wantErr {
|
||||
return
|
||||
}
|
||||
|
||||
var got SendEmailPayload
|
||||
if err := json.Unmarshal(gotPayload, &got); err != nil {
|
||||
t.Fatalf("json.Unmarshal(%s) error = %v", gotPayload, err)
|
||||
}
|
||||
if diff := cmp.Diff(tt.want, got); diff != "" {
|
||||
t.Errorf("ValidatePayload(%s) mismatch (-want +got):\n%s", tt.payload, diff)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`Execute` 测试优先验证业务服务调用、错误返回和结果摘要；日志可只验证关键路径，避免把精确日志文本写成脆弱断言。
|
||||
|
||||
## Admin Dispatch 测试形状
|
||||
|
||||
Admin dispatch 测试关注通用链路是否调用了 `PayloadValidator`，不要为每种任务在 handler 里写 if 分支。
|
||||
|
||||
```go
|
||||
func TestDispatchTaskValidatesPayload(t *testing.T) {
|
||||
// 1. 初始化测试 DB 和 task.AsynqClient。
|
||||
// 2. 注册测试 handler: task.RegisterHandler(task.SendEmailTask, &user.SendEmailHandler{})
|
||||
// 3. POST /api/v1/admin/tasks/dispatch，传入非法 payload。
|
||||
// 4. 断言响应为 400，错误信息清晰，且没有创建可执行任务。
|
||||
}
|
||||
```
|
||||
|
||||
需要 Redis/Asynq 时优先复用项目现有测试模式；没有现成依赖时可用 `miniredis` 初始化 `task.AsynqClient`。不要把 `internal/task` 依赖塞进通用 testhelper 造成 import cycle。
|
||||
Reference in New Issue
Block a user