tcjlj/OpenFlare
1
0
Fork 0
mirror of https://github.com/Rain-kl/OpenFlare.git synced 2026-10-10 17:26:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

docs(agents): update AGENTS.md and development skills for cordis architecture

Browse Source
This commit is contained in:
ryan
2026-08-27 23:56:16 +08:00
parent 94c5aa4cd3
commit 75f226cfbf
12 changed files with 660 additions and 977 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.agents/skills/new-async-task/SKILL.md
+124 -88
View File
@@ -1,121 +1,157 @@
---
name: "new-async-task"
description: "Wavelet 项目专用：新增或修改 Asynq 异步任务、后台任务、定时任务、任务元数据、TaskHandler、TaskParam、PayloadValidator、AppendLog、任务重试、任务执行记录或 Admin 任务 API 时必须使用。"
description: "Wavelet 项目专用：新增或修改基于 Cordis 插件的 Asynq 异步任务、后台 Worker 消费处理器、Cron 定时调度任务与任务执行追踪时必须使用。"
---
# 异步任务开发
# 异步任务与定时调度开发规范 (Cordis 插件化架构)
开始前阅读根目录 `AGENTS.md`。只修改任务相关链路，遵守项目路由、日志、数据库迁移和质量门禁要求。
本技能是 Wavelet 在 Cordis 微内核与插件化架构下，进行 Asynq 异步后台任务与 Cron 定时调度开发的唯一指导规范。
## 开始前
---
按任务范围检查当前实现：
## 1. 核心架构：插件内自包含任务声明
- `internal/infra/task/handler.go`：`TaskHandler`、`TaskResult`、`PayloadValidator`
- `internal/infra/task/meta.go`：`TaskMeta`、`TaskParam`
- `internal/infra/task/executor.go`：下发、执行、日志、重试、`OnTaskCompleted` 订阅
- `internal/infra/task/handlers/register.go`：Handler 和元数据注册（由 bootstrap 调用）
- `internal/platform/bootstrap/bootstrap.go`：任务注册与进程级装配入口
- `internal/infra/task/worker/worker.go`：Worker 路由和队列
- `internal/infra/task/scheduler/scheduler.go`：定时调度
- `internal/apps/admin/task/routers.go`：Admin 任务 API
- `internal/model/task_execution.go`：执行记录和日志持久化
在 Cordis 架构中，后台 Worker 消费与定时调度**不再集中在中心化的注册表**，而是由各个业务插件在自身的 `Apply` 方法中通过微内核扩展点直接声明。
需要模板时阅读 [references/CODE-EXAMPLES.md](references/CODE-EXAMPLES.md)。
### 扩展点矩阵
## 实现要求
| 扩展点方法 | 说明 | 适用场景 |
| :--- | :--- | :--- |
| `ctx.Task().Register(pattern, handler, opts...)` | 注册 Asynq 任务类型与消费处理器 | 异步耗时计算、队列任务、通知外发 |
| `ctx.Schedule().RegisterCron(spec, taskType, payload)` | 注册 Cron 表达式定时调度任务 | 周期统计、定时清理、健康检查 |
### 任务定义
---
- 在 `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` 只描述前端表单，不代替服务端校验。
## 2. 异步任务开发全流程
### Handler
### 步骤 1：定义任务 Payload 结构与类型常量
- Handler 必须实现 `task.TaskHandler`。
- 有参数任务必须实现 `task.PayloadValidator`，负责校验和标准化 Admin 下发参数。
- `Execute` 必须再次解析 payload；不要假设入口一定经过 Admin 校验。
- 成功返回 `&task.TaskResult{Message: ..., Detail: ...}`。
- 失败返回 error，由任务框架处理状态和重试。
- 不要吞掉关键错误。
- 复杂 SQL 放到 `internal/model/` 或模块内的业务服务层（如 `internal/apps/<module>/service.go` 或 `logics.go`）。
在插件内（如 `plugins/domain/order/tasks.go`）：
### 注册
```go
package order
- 在 `internal/infra/task/handlers/register.go` 同时注册 Handler 和 `TaskMeta`。
- 不要在其他位置单独注册任务。
- **禁止**在业务包 `routers.go` 或 `init()` 中调用 `task.RegisterHandler`；统一由 `bootstrap.RegisterTasks()` → `taskhandlers.Register()` 在进程启动时装配。
- 任务完成钩子（如 push 通知）通过 `task.OnTaskCompleted` 注册，在 `bootstrap.RegisterTaskListeners()` 中装配（Worker/`all` 进程）。
import (
"context"
"encoding/json"
"time"
### 进程装配分工
"github.com/hibiken/asynq"
)
| 进程 | 注册入口 |
| :--- | :--- |
| `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()` |
const (
TaskTypeOrderTimeoutCancel = "order:timeout_cancel"
)
所有 `Register*` 使用 `sync.Once`，重复调用安全。
// OrderTimeoutPayload 定义任务入参
type OrderTimeoutPayload struct {
OrderID string `json:"order_id"`
Reason string `json:"reason"`
CreatedAt int64 `json:"created_at"`
}
```
### 测试
### 步骤 2：实现任务执行处理器 (Handler)
- 依赖已注册任务类型或 Handler 的测试（如 `internal/apps/admin/task/routers_test.go`），必须在 setup 中显式调用 `bootstrap.RegisterTasks()`。
- 不得依赖 `init()` 副作用或 import 链触发注册。
Handler 必须接受 `ctx context.Context, t *asynq.Task`，返回 `error`：
## 日志要求
```go
func (p *Plugin) handleOrderTimeoutCancel(ctx context.Context, t *asynq.Task) error {
var payload OrderTimeoutPayload
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return err // 反序列化失败，直接中断
}
- 在 `TaskHandler.Execute` 中使用 `task.AppendLog(ctx, format, args...)`。
- 记录任务开始、参数摘要、批次进度、关键状态、可继续错误和完成摘要。
- 批量处理按批次记录；禁止为大循环中的每条数据写日志。
- 不要直接修改任务日志的 Redis key 或 `w_task_executions.log`。
// 记录任务日志
// task.AppendLog(ctx, "开始处理订单超时关单: order_id=%s", payload.OrderID)
日志框架约束：
// 执行业务逻辑
if err := p.svc.CancelTimeoutOrder(ctx, payload.OrderID, payload.Reason); err != nil {
// 返回 error 触发 Asynq 框架自动重试
return err
}
- 执行状态实时写入数据库：`pending`、`running`、`succeeded`、`failed`。
- 实时日志写入 Redis，每个任务最多保留最近 1000 行。
- Redis 日志 TTL 为 24 小时，每次追加时刷新。
- 查询时优先返回 Redis 日志，Redis 不存在时读取数据库。
- 任务成功或自动重试耗尽后，将日志写入数据库并删除 Redis 缓冲。
- 自动重试期间保留同一 taskID 的 Redis 日志。
return nil
}
```
## 重试要求
### 步骤 3：在插件 `Apply` 中注册任务与定时调度
- Handler 返回 error 以触发 Asynq 自动重试。
- 不要在 Handler 内自行实现重复重试循环。
- Admin 手动重试只允许：
- 原任务状态为 `failed`
- `Retryable=true`
- `RetryCount < MaxRetry`
- 修改重试行为时同时检查：
- `internal/infra/task/executor.go`
- `internal/model/task_execution.go`
- `internal/apps/admin/task/routers.go`
- 前端任务执行列表
```go
func (p *Plugin) Apply(ctx *core.Context) error {
// 1. 注册异步任务处理器
ctx.Task().Register(
TaskTypeOrderTimeoutCancel,
p.handleOrderTimeoutCancel,
extpoints.WithTaskRetry(3),
extpoints.WithTaskTimeout(5*time.Minute),
)
## 定时任务
// 2. 注册定时调度任务 (例如每天凌晨 2 点执行汇总)
ctx.Schedule().RegisterCron(
"0 2 * * *",
"order:daily_settlement",
map[string]any{"scope": "all"},
)
- 默认定时任务必须通过 Goose SQL 迁移写入 `schedules`。
- PostgreSQL 和 SQLite 迁移必须同时提供。
- 初始化 SQL 必须幂等。
- 涉及迁移时使用 `database-migration` skill。
return nil
}
```
## Admin API
### 步骤 4：在业务逻辑中投递异步任务
- Handler 放在现有 Admin task 模块或 `internal/apps/admin/<module>/`。
- 路由只在 `internal/router/router.go` 注册。
- 响应保持 `{ "error_msg": "", "data": ... }`。
- 分页数据保持 `{ "total": 0, "results": [] }`。
- Swagger 注释必须完整；API 变化后运行 `make swagger`。
当业务需要下发延迟或异步任务时：
## 前端
```go
func (s *OrderService) EnqueueTimeoutCheck(ctx context.Context, orderID string) error {
payloadBytes, _ := json.Marshal(OrderTimeoutPayload{
OrderID: orderID,
Reason: "15分钟未支付自动关单",
CreatedAt: time.Now().Unix(),
})
- 仅任务元数据变化时，优先复用现有动态任务表单，不新增页面。
- API 调用必须通过 `frontend/lib/services/`。
- 修改 shadcn/ui 时使用 `shadcn` skill。
- 不使用 `any`。
- 页面根容器使用 `w-full`，不添加页面级 `max-w-*`。
task := asynq.NewTask(
TaskTypeOrderTimeoutCancel,
payloadBytes,
asynq.ProcessIn(15*time.Minute), // 延迟 15 分钟执行
asynq.MaxRetry(3),
)
// 投递到任务客户端
_, err := s.taskClient.EnqueueContext(ctx, task)
return err
}
```
---
## 3. 运行切面透明性 (Profile Transparency)
Cordis 微内核支持多种启动切面（`api`、`worker`、`schedule`、`all`）：
- 插件开发者**无需在插件代码中编写 `if mode == "worker"` 分支**。
- 插件只需在 `Apply` 中把任务与调度注册进 `Context`。
- 当进程以 `worker` 切面启动时，微内核的 `driver_asynq_worker` 驱动会自动拾取并监听已注册的任务。
- 当进程以 `schedule` 切面启动时，`driver_asynq_cron` 驱动会自动启动调度器引擎。
---
## 4. 任务日志与重试规范
1. **日志记录**：
- 记录任务启动参数摘要、分批处理进度及最终完成统计。
- 大循环处理中应按批次记录日志，禁止每条数据单独打日志刷屏。
2. **重试机制**：
- Handler 返回 error 即自动触发 Asynq 重试策略。
- 禁止在 Handler 内部编写裸 `for` 死循环重试。
3. **幂等性保障**：
- 任务由于网络波动或超时可能被重复消费，业务操作必须实现幂等保护（如基于订单状态机检查或分布式锁 `ctx.DistLock()`）。
---
## 5. 质量验证
```bash
make format
make code-check
go test ./plugins/...
```
.agents/skills/new-async-task/references/CODE-EXAMPLES.md
+111 -234
View File
@@ -1,277 +1,154 @@
# Wavelet 异步任务代码示例
# Wavelet 异步任务代码示例 (Cordis 插件化架构)
这些示例用于新增或修改 Wavelet Asynq 任务时快速套用。复制前先对照当前代码，因为任务框架可能随项目演进。
这些示例用于在 Cordis 插件中开发或修改 Asynq 任务时快速套用。
## 任务元数据与常量定义
---
在对应的业务包 `internal/apps/<module>/tasks.go` 中定义 Asynq task type、Admin task type 和 `TaskMeta`。
## 1. 任务定义与 Handler 编写
```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 校验路径。
在对应的业务插件中（如 `plugins/domain/user/tasks.go`）：
```go
package user
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"github.com/Rain-kl/Wavelet/internal/task"
"github.com/hibiken/asynq"
)
// 异步任务类型标识。格式推荐为 "{plugin}:{action}"
const TaskTypeSendEmail = "user:send_email"
type SendEmailPayload struct {
To string `json:"to"`
Subject string `json:"subject"`
Body string `json:"body"`
To string `json:"to"`
Subject string `json:"subject"`
Body string `json:"body"`
}
type SendEmailHandler struct{}
// Handler 处理函数
func (p *Plugin) handleSendEmail(ctx context.Context, t *asynq.Task) error {
var req SendEmailPayload
if err := json.Unmarshal(t.Payload(), &req); err != nil {
return fmt.Errorf("解析任务参数: %w", err)
}
func (h *SendEmailHandler) ValidatePayload(payload []byte) ([]byte, error) {
if len(payload) == 0 {
return nil, errors.New("任务参数不能为空")
}
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 errors.New("to、subject、body 不能为空")
}
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
// 执行实际邮件发送业务逻辑
return p.emailSvc.Send(ctx, req.To, req.Subject, req.Body)
}
```
## 统一注册
---
在 `internal/infra/task/handlers/register.go` 注册。Admin dispatch 的 `ValidateAndNormalizePayload` 和 Worker 执行都依赖这里。
## 2. 插件内自包含注册 (`Apply`)
在插件的 `Apply(ctx *core.Context)` 中：
```go
package handlers
package user
import (
"github.com/Rain-kl/Wavelet/internal/apps/upload"
"github.com/Rain-kl/Wavelet/internal/apps/user"
"github.com/Rain-kl/Wavelet/internal/task"
"time"
"github.com/Rain-kl/Wavelet/core"
"github.com/Rain-kl/Wavelet/core/extpoints"
)
func Register() {
task.RegisterHandler(task.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{})
task.RegisterHandler(task.SendEmailTask, &user.SendEmailHandler{})
func (p *Plugin) Apply(ctx *core.Context) error {
// 1. 注册 Asynq 异步任务消费处理器
ctx.Task().Register(
TaskTypeSendEmail,
p.handleSendEmail,
extpoints.WithTaskRetry(3),
extpoints.WithTaskTimeout(2*time.Minute),
)
// 2. 注册 Cron 调度任务（例如每天凌晨 3 点清理过期 Token）
ctx.Schedule().RegisterCron(
"0 3 * * *",
"user:cleanup_expired_tokens",
map[string]any{"scope": "expired"},
)
return nil
}
```
## Cron 调度和配置
---
系统默认的定时任务必须通过 Goose SQL 迁移初始化插入到 `schedules` 表。
在 `internal/infra/persistence/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、缺失必填和标准化。
## 3. 业务中投递异步任务
```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,
},
}
func (s *UserService) TriggerWelcomeEmail(ctx context.Context, toEmail, username string) error {
payload, _ := json.Marshal(SendEmailPayload{
To: toEmail,
Subject: "欢迎加入",
Body: fmt.Sprintf("你好 %s，欢迎使用我们的平台！", username),
})
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
}
task := asynq.NewTask(
TaskTypeSendEmail,
payload,
asynq.MaxRetry(3),
)
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)
}
})
}
_, err := s.taskClient.EnqueueContext(ctx, task)
return err
}
```
`Execute` 测试优先验证业务服务调用、错误返回和结果摘要；日志可只验证关键路径，避免把精确日志文本写成脆弱断言。
---
## Admin Dispatch 测试形状
Admin dispatch 测试关注通用链路是否调用了 `PayloadValidator`，不要为每种任务在 handler 里写 if 分支。
## 4. 任务处理函数单元测试
```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，错误信息清晰，且没有创建可执行任务。
func TestSendEmailPayloadValidation(t *testing.T) {
tests := []struct {
name string
payload []byte
wantErr bool
}{
{
name: "valid payload",
payload: []byte(`{"to":"user@example.com","subject":"hi","body":"welcome"}`),
wantErr: false,
},
{
name: "empty payload",
payload: nil,
wantErr: true,
},
{
name: "missing required fields",
payload: []byte(`{"to":"user@example.com","subject":""}`),
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var req SendEmailPayload
err := json.Unmarshal(tt.payload, &req)
if err == nil {
if strings.TrimSpace(req.To) == "" || strings.TrimSpace(req.Subject) == "" || strings.TrimSpace(req.Body) == "" {
err = errors.New("missing fields")
}
}
if (err != nil) != tt.wantErr {
t.Fatalf("validation error = %v, wantErr = %v", err, tt.wantErr)
}
})
}
}
```
需要 Redis/Asynq 时优先复用项目现有测试模式；没有现成依赖时可用 `miniredis` 初始化 `task.AsynqClient`。不要把 `internal/infra/task` 依赖塞进通用 testhelper 造成 import cycle。