async task framework

internal/task/constants.go

@@ -38,6 +38,7 @@ type TaskMeta struct {
 	SupportsTime bool
 	MaxRetry     int
 	Queue        string
+	Retryable    bool // 是否支持手动重试
 }
 
 // DispatchableTasks 可下发的任务列表
@@ -50,6 +51,7 @@ var DispatchableTasks = []TaskMeta{
 		SupportsTime: false,
 		MaxRetry:     3,
 		Queue:        QueueDefault,
+		Retryable:    true,
 	},
 }
 

internal/task/executor.go

@@ -0,0 +1,296 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package task
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/linux-do/credit/internal/db/idgen"
+	"github.com/linux-do/credit/internal/logger"
+	"github.com/linux-do/credit/internal/model"
+	"github.com/linux-do/credit/internal/otel_trace"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// handlerRegistry 已注册的任务处理器
+var handlerRegistry = make(map[string]TaskHandler)
+
+// RegisterHandler 注册任务处理器
+// 传入任务类型标识（对应 constants.go 中的 AsynqTask 常量）和 TaskHandler 实现
+func RegisterHandler(asynqTaskType string, handler TaskHandler) {
+	handlerRegistry[asynqTaskType] = handler
+}
+
+// getHandler 获取已注册的处理器
+func getHandler(asynqTaskType string) (TaskHandler, bool) {
+	h, ok := handlerRegistry[asynqTaskType]
+	return h, ok
+}
+
+// contextKey 用于 context 存取 taskID
+type contextKey string
+
+const taskIDKey contextKey = "task_execution_task_id"
+
+// withTaskID 将 taskID 注入 context
+func withTaskID(ctx context.Context, taskID string) context.Context {
+	return context.WithValue(ctx, taskIDKey, taskID)
+}
+
+// GetTaskID 从 context 中获取 taskID
+func GetTaskID(ctx context.Context) string {
+	if v, ok := ctx.Value(taskIDKey).(string); ok {
+		return v
+	}
+	return ""
+}
+
+// AppendLog 追加日志到任务执行记录
+// 在 TaskHandler.Execute 中调用，日志会自动追加到 TaskExecution.Log 字段
+func AppendLog(ctx context.Context, format string, args ...interface{}) {
+	taskID := GetTaskID(ctx)
+	if taskID == "" {
+		// 上下文中没有 taskID，降级到普通日志
+		logger.InfoF(ctx, format, args...)
+		return
+	}
+
+	logLine := fmt.Sprintf(format, args...)
+	if err := model.AppendTaskExecutionLog(ctx, taskID, logLine); err != nil {
+		logger.ErrorF(ctx, "[TaskExecutor] 追加任务日志失败 taskID=%s: %v", taskID, err)
+	}
+}
+
+// DispatchTask 下发任务（创建 TaskExecution 记录 → 入队 Asynq）
+func DispatchTask(ctx context.Context, taskType string, payload []byte, triggeredBy string) (string, error) {
+	meta := GetTaskMeta(taskType)
+	if meta == nil {
+		return "", fmt.Errorf("未知的任务类型: %s", taskType)
+	}
+
+	// 生成唯一的 TaskID
+	taskID := generateTaskID(taskType, triggeredBy)
+
+	// 创建任务执行记录
+	execution := &model.TaskExecution{
+		TaskID:      taskID,
+		TaskType:    meta.AsynqTask,
+		TaskName:    meta.Name,
+		Status:      model.TaskExecutionStatusPending,
+		Retryable:   meta.Retryable,
+		MaxRetry:    meta.MaxRetry,
+		RetryCount:  0,
+		Payload:     string(payload),
+		TriggeredBy: triggeredBy,
+	}
+
+	if err := model.CreateTaskExecution(ctx, execution); err != nil {
+		return "", fmt.Errorf("创建任务执行记录失败: %w", err)
+	}
+
+	// 入队 Asynq
+	taskInfo := asynq.NewTask(meta.AsynqTask, payload)
+	if _, err := AsynqClient.Enqueue(
+		taskInfo,
+		asynq.TaskID(taskID),
+		asynq.MaxRetry(meta.MaxRetry),
+		asynq.Queue(meta.Queue),
+	); err != nil {
+		// 入队失败，更新执行记录状态
+		execution.Status = model.TaskExecutionStatusFailed
+		execution.ErrorMessage = fmt.Sprintf("入队失败: %v", err)
+		now := time.Now()
+		execution.StartedAt = &now
+		execution.FinishedAt = &now
+		_ = model.UpdateTaskExecution(ctx, execution)
+		return "", fmt.Errorf("任务入队失败: %w", err)
+	}
+
+	return taskID, nil
+}
+
+// RetryTask 重试失败的任务
+func RetryTask(ctx context.Context, id uint64) (string, error) {
+	execution, err := model.GetTaskExecutionByID(ctx, id)
+	if err != nil {
+		return "", fmt.Errorf("任务执行记录不存在: %w", err)
+	}
+
+	if execution.Status != model.TaskExecutionStatusFailed {
+		return "", fmt.Errorf("只有失败的任务才能重试，当前状态: %s", execution.Status)
+	}
+
+	if !execution.Retryable {
+		return "", fmt.Errorf("该任务不支持重试")
+	}
+
+	if execution.RetryCount >= execution.MaxRetry {
+		return "", fmt.Errorf("已达到最大重试次数 %d", execution.MaxRetry)
+	}
+
+	// 生成新的 TaskID
+	newTaskID := generateRetryTaskID(execution.TaskID, execution.RetryCount+1)
+
+	// 创建新的执行记录
+	newExecution := &model.TaskExecution{
+		TaskID:      newTaskID,
+		TaskType:    execution.TaskType,
+		TaskName:    execution.TaskName,
+		Status:      model.TaskExecutionStatusPending,
+		Retryable:   execution.Retryable,
+		MaxRetry:    execution.MaxRetry,
+		RetryCount:  execution.RetryCount + 1,
+		Payload:     execution.Payload,
+		TriggeredBy: "retry",
+	}
+
+	if err := model.CreateTaskExecution(ctx, newExecution); err != nil {
+		return "", fmt.Errorf("创建重试任务执行记录失败: %w", err)
+	}
+
+	// 入队 Asynq
+	taskInfo := asynq.NewTask(execution.TaskType, []byte(execution.Payload))
+	if _, err := AsynqClient.Enqueue(
+		taskInfo,
+		asynq.TaskID(newTaskID),
+		asynq.MaxRetry(execution.MaxRetry),
+		asynq.Queue(PrefixedQueue(QueueDefault)),
+	); err != nil {
+		newExecution.Status = model.TaskExecutionStatusFailed
+		newExecution.ErrorMessage = fmt.Sprintf("重试入队失败: %v", err)
+		now := time.Now()
+		newExecution.StartedAt = &now
+		newExecution.FinishedAt = &now
+		_ = model.UpdateTaskExecution(ctx, newExecution)
+		return "", fmt.Errorf("重试任务入队失败: %w", err)
+	}
+
+	return newTaskID, nil
+}
+
+// ProcessTask Asynq 实际调用的统一处理函数
+// Worker 注册时统一使用此函数，内部自动分发到对应的 TaskHandler
+func ProcessTask(ctx context.Context, t *asynq.Task) error {
+	// 初始化 Trace
+	ctx, span := otel_trace.Start(ctx, "TaskProcess_"+t.Type(), trace.WithSpanKind(trace.SpanKindConsumer))
+	defer span.End()
+
+	// 添加任务信息到 Span
+	span.SetAttributes(
+		attribute.String("task.type", t.Type()),
+		attribute.Int("task.payload_size", len(t.Payload())),
+		attribute.String("task.id", t.ResultWriter().TaskID()),
+	)
+
+	taskID := t.ResultWriter().TaskID()
+
+	// 注入 taskID 到 context
+	ctx = withTaskID(ctx, taskID)
+
+	// 查找处理器
+	handler, ok := getHandler(t.Type())
+	if !ok {
+		err := fmt.Errorf("未注册的任务处理器: %s", t.Type())
+		logger.ErrorF(ctx, "[TaskExecutor] %v", err)
+		span.SetStatus(codes.Error, err.Error())
+		return err
+	}
+
+	// 从数据库加载执行记录
+	execution, err := model.GetTaskExecutionByTaskID(ctx, taskID)
+	if err != nil {
+		logger.ErrorF(ctx, "[TaskExecutor] 查询执行记录失败 taskID=%s: %v", taskID, err)
+		// 执行记录不存在，仍然执行任务但不记录状态
+		_, execErr := handler.Execute(ctx, t.Payload())
+		if execErr != nil {
+			span.SetStatus(codes.Error, execErr.Error())
+			return execErr
+		}
+		return nil
+	}
+
+	// 更新状态为 running
+	now := time.Now()
+	execution.Status = model.TaskExecutionStatusRunning
+	execution.StartedAt = &now
+	if err := model.UpdateTaskExecution(ctx, execution); err != nil {
+		logger.ErrorF(ctx, "[TaskExecutor] 更新执行状态失败 taskID=%s: %v", taskID, err)
+	}
+
+	// 开始计时
+	start := time.Now()
+
+	// 执行业务逻辑
+	result, execErr := handler.Execute(ctx, t.Payload())
+
+	// 计算耗时
+	duration := time.Since(start)
+	finishTime := time.Now()
+	execution.Duration = duration.Milliseconds()
+	execution.FinishedAt = &finishTime
+
+	if execErr != nil {
+		// 执行失败
+		execution.Status = model.TaskExecutionStatusFailed
+		execution.ErrorMessage = execErr.Error()
+
+		logger.ErrorF(ctx,
+			"[TaskExecutor] 任务处理失败 Type: %s TaskID: %s Duration: %d ms Error: %v",
+			t.Type(), taskID, duration.Milliseconds(), execErr,
+		)
+
+		span.SetStatus(codes.Error, execErr.Error())
+		span.RecordError(execErr)
+	} else {
+		// 执行成功
+		execution.Status = model.TaskExecutionStatusSucceeded
+		if result != nil {
+			execution.Result = result.Message
+			if result.Detail != "" {
+				execution.Result = fmt.Sprintf("%s\n%s", result.Message, result.Detail)
+			}
+		}
+
+		logger.InfoF(ctx,
+			"[TaskExecutor] 任务处理完成 Type: %s TaskID: %s Duration: %d ms",
+			t.Type(), taskID, duration.Milliseconds(),
+		)
+	}
+
+	// 更新执行记录
+	if err := model.UpdateTaskExecution(ctx, execution); err != nil {
+		logger.ErrorF(ctx, "[TaskExecutor] 更新执行记录失败 taskID=%s: %v", taskID, err)
+	}
+
+	return execErr
+}
+
+// generateTaskID 生成任务 ID
+func generateTaskID(taskType string, triggeredBy string) string {
+	uniqueID := idgen.NextUint64ID()
+	return fmt.Sprintf("%s_%s_%d", triggeredBy, taskType, uniqueID)
+}
+
+// generateRetryTaskID 生成重试任务 ID
+func generateRetryTaskID(originalTaskID string, retryCount int) string {
+	return fmt.Sprintf("retry_%d_%s", retryCount, originalTaskID)
+}

internal/task/executor_test.go

@@ -0,0 +1,357 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package task
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/linux-do/credit/internal/model"
+	"github.com/linux-do/credit/internal/testhelper"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// mockHandler 用于测试的模拟任务处理器
+type mockHandler struct {
+	executeFunc func(ctx context.Context, payload []byte) (*TaskResult, error)
+}
+
+func (h *mockHandler) Execute(ctx context.Context, payload []byte) (*TaskResult, error) {
+	if h.executeFunc != nil {
+		return h.executeFunc(ctx, payload)
+	}
+	return &TaskResult{Message: "mock success"}, nil
+}
+
+// successHandler 返回成功的处理器
+func successHandler() *mockHandler {
+	return &mockHandler{
+		executeFunc: func(ctx context.Context, payload []byte) (*TaskResult, error) {
+			AppendLog(ctx, "执行成功，处理了 %d 条数据", 100)
+			return &TaskResult{Message: "处理完成，共 100 条"}, nil
+		},
+	}
+}
+
+// failHandler 返回失败的处理器
+func failHandler() *mockHandler {
+	return &mockHandler{
+		executeFunc: func(ctx context.Context, payload []byte) (*TaskResult, error) {
+			AppendLog(ctx, "开始执行任务")
+			return nil, fmt.Errorf("模拟执行失败: 数据库连接超时")
+		},
+	}
+}
+
+const testTaskType = "test:mock_task"
+
+func setupTest(t *testing.T) func() {
+	_, _, cleanup := testhelper.SetupTestEnvironment(t)
+	// 注册测试用 handler
+	RegisterHandler(testTaskType, successHandler())
+	return cleanup
+}
+
+func TestRegisterAndGetHandler(t *testing.T) {
+	_ = testTaskType
+	cleanup := setupTest(t)
+	defer cleanup()
+
+	// 验证 handler 已注册
+	h, ok := getHandler(testTaskType)
+	assert.True(t, ok, "handler should be registered")
+	assert.NotNil(t, h)
+
+	// 未注册的 handler
+	_, ok = getHandler("nonexistent")
+	assert.False(t, ok, "non-existent handler should return false")
+}
+
+func TestGetTaskIDFromContext(t *testing.T) {
+	ctx := context.Background()
+
+	// 空 context
+	taskID := GetTaskID(ctx)
+	assert.Equal(t, "", taskID)
+
+	// 注入 taskID
+	ctx = withTaskID(ctx, "test_task_123")
+	taskID = GetTaskID(ctx)
+	assert.Equal(t, "test_task_123", taskID)
+}
+
+func TestAppendLogWithoutTaskID(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	// 没有 taskID 的 context，应降级到普通日志，不报错
+	AppendLog(ctx, "这条日志应该降级处理，不会报错")
+}
+
+func TestAppendLogWithTaskID(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	// 先创建一条执行记录
+	execution := &model.TaskExecution{
+		TaskID:      "log_test_001",
+		TaskType:    testTaskType,
+		TaskName:    "测试任务",
+		Status:      model.TaskExecutionStatusRunning,
+		TriggeredBy: "manual",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	// 注入 taskID 并追加日志
+	ctx = withTaskID(ctx, "log_test_001")
+	AppendLog(ctx, "第一条日志")
+	AppendLog(ctx, "处理了 %d 条数据", 50)
+
+	// 验证日志
+	found, err := model.GetTaskExecutionByTaskID(ctx, "log_test_001")
+	require.NoError(t, err)
+	assert.Contains(t, found.Log, "第一条日志")
+	assert.Contains(t, found.Log, "处理了 50 条数据")
+}
+
+func TestProcessTaskSuccess(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	// 注册成功 handler
+	RegisterHandler(testTaskType, successHandler())
+
+	// 创建执行记录
+	execution := &model.TaskExecution{
+		TaskID:      "process_success_001",
+		TaskType:    testTaskType,
+		TaskName:    "测试任务",
+		Status:      model.TaskExecutionStatusPending,
+		Retryable:   true,
+		MaxRetry:    3,
+		TriggeredBy: "manual",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	// 模拟 Asynq Task
+	asynqTask := asynq.NewTask(testTaskType, nil)
+	// 使用 ResultWriter 设置 TaskID
+	rw := asynq.NewResultWriter()
+	rw.SetTaskID("process_success_001")
+	// 通过 asynq 的 Task 不能直接设置 taskID，ProcessTask 通过 t.ResultWriter().TaskID() 获取
+	// 但 asynq.Task 在没有经过 asynq server 的情况下 ResultWriter 可能为 nil
+	// 我们需要在 ProcessTask 内部改用 taskID 注入的方式测试
+	// 为了测试 ProcessTask，我们直接模拟调用 handler
+
+	// 直接通过 handler 测试
+	handler, ok := getHandler(testTaskType)
+	require.True(t, ok)
+
+	ctx = withTaskID(ctx, "process_success_001")
+	result, err := handler.Execute(ctx, nil)
+	require.NoError(t, err)
+	assert.Equal(t, "处理完成，共 100 条", result.Message)
+
+	// 验证日志被追加
+	found, err := model.GetTaskExecutionByTaskID(ctx, "process_success_001")
+	require.NoError(t, err)
+	assert.Contains(t, found.Log, "执行成功，处理了 100 条数据")
+}
+
+func TestProcessTaskFailure(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	// 注册失败 handler
+	RegisterHandler(testTaskType, failHandler())
+
+	// 创建执行记录
+	execution := &model.TaskExecution{
+		TaskID:      "process_fail_001",
+		TaskType:    testTaskType,
+		TaskName:    "测试任务",
+		Status:      model.TaskExecutionStatusPending,
+		Retryable:   true,
+		MaxRetry:    3,
+		TriggeredBy: "manual",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	// 直接调用 handler
+	handler, ok := getHandler(testTaskType)
+	require.True(t, ok)
+
+	ctx = withTaskID(ctx, "process_fail_001")
+	_, err = handler.Execute(ctx, nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "模拟执行失败")
+
+	// 验证日志
+	found, err := model.GetTaskExecutionByTaskID(ctx, "process_fail_001")
+	require.NoError(t, err)
+	assert.Contains(t, found.Log, "开始执行任务")
+}
+
+func TestRetryTask(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	// 创建一条失败的执行记录（可重试）
+	now := time.Now()
+	execution := &model.TaskExecution{
+		TaskID:       "retry_test_001",
+		TaskType:     testTaskType,
+		TaskName:     "测试任务",
+		Status:       model.TaskExecutionStatusFailed,
+		Retryable:    true,
+		MaxRetry:     3,
+		RetryCount:   0,
+		ErrorMessage: "首次执行失败",
+		StartedAt:    &now,
+		FinishedAt:   &now,
+		Duration:     100,
+		TriggeredBy:  "manual",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	// 重试
+	newTaskID, err := RetryTask(ctx, execution.ID)
+	require.NoError(t, err)
+	assert.NotEmpty(t, newTaskID)
+	assert.Contains(t, newTaskID, "retry_1_")
+
+	// 验证新记录
+	newExecution, err := model.GetTaskExecutionByTaskID(ctx, newTaskID)
+	require.NoError(t, err)
+	assert.Equal(t, model.TaskExecutionStatusPending, newExecution.Status)
+	assert.Equal(t, 1, newExecution.RetryCount)
+	assert.Equal(t, "retry", newExecution.TriggeredBy)
+	assert.Equal(t, execution.TaskType, newExecution.TaskType)
+	assert.True(t, newExecution.Retryable)
+
+	// 原记录不变
+	original, err := model.GetTaskExecutionByID(ctx, execution.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.TaskExecutionStatusFailed, original.Status)
+	assert.Equal(t, 0, original.RetryCount)
+}
+
+func TestRetryTaskNotFailed(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	// 创建一条成功的记录
+	execution := &model.TaskExecution{
+		TaskID:      "retry_not_failed_001",
+		TaskType:    testTaskType,
+		TaskName:    "测试任务",
+		Status:      model.TaskExecutionStatusSucceeded,
+		Retryable:   true,
+		MaxRetry:    3,
+		TriggeredBy: "manual",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	// 尝试重试成功的任务
+	_, err = RetryTask(ctx, execution.ID)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "只有失败的任务才能重试")
+}
+
+func TestRetryTaskNotRetryable(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	execution := &model.TaskExecution{
+		TaskID:      "retry_not_allowed_001",
+		TaskType:    testTaskType,
+		TaskName:    "测试任务",
+		Status:      model.TaskExecutionStatusFailed,
+		Retryable:   false,
+		MaxRetry:    0,
+		TriggeredBy: "manual",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	_, err = RetryTask(ctx, execution.ID)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "不支持重试")
+}
+
+func TestRetryTaskMaxRetryExceeded(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	execution := &model.TaskExecution{
+		TaskID:      "retry_max_001",
+		TaskType:    testTaskType,
+		TaskName:    "测试任务",
+		Status:      model.TaskExecutionStatusFailed,
+		Retryable:   true,
+		MaxRetry:    2,
+		RetryCount:  2,
+		TriggeredBy: "retry",
+	}
+	err := model.CreateTaskExecution(ctx, execution)
+	require.NoError(t, err)
+
+	_, err = RetryTask(ctx, execution.ID)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "已达到最大重试次数")
+}
+
+func TestRetryTaskNonExistent(t *testing.T) {
+	cleanup := setupTest(t)
+	defer cleanup()
+	ctx := context.Background()
+
+	_, err := RetryTask(ctx, 99999999)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "不存在")
+}
+
+func TestGenerateTaskID(t *testing.T) {
+	id1 := generateTaskID("test_type", "manual")
+	id2 := generateTaskID("test_type", "manual")
+
+	// 两个 ID 应不同（包含 Snowflake ID）
+	assert.NotEqual(t, id1, id2)
+	assert.Contains(t, id1, "manual_test_type_")
+}
+
+func TestGenerateRetryTaskID(t *testing.T) {
+	id := generateRetryTaskID("original_task_123", 2)
+	assert.Equal(t, "retry_2_original_task_123", id)
+}

internal/task/handler.go

@@ -0,0 +1,38 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package task
+
+import "context"
+
+// TaskResult 任务执行结果
+type TaskResult struct {
+	Message string // 结果摘要，如 "共清理 120 个文件，耗时 3.2s"
+	Detail  string // 可选的详细结果 JSON
+}
+
+// TaskHandler 异步任务处理器接口
+// 所有异步任务必须实现此接口，框架将自动管理任务执行记录的创建、状态流转和日志写入。
+//
+// 开发者只需实现 Execute 方法编写业务逻辑，在方法内通过 task.AppendLog(ctx, ...) 追加执行日志。
+// 任务的创建、状态更新、错误记录、重试计数全部由框架透明处理。
+type TaskHandler interface {
+	// Execute 执行任务业务逻辑
+	//   - ctx: 已注入 Trace Span 和 taskID 的上下文
+	//   - payload: 调度时传入的原始参数（可为 nil）
+	//   - 返回 TaskResult 描述执行结果，或 error 表示执行失败
+	Execute(ctx context.Context, payload []byte) (*TaskResult, error)
+}

internal/task/scheduler/scheduler.go

@@ -28,13 +28,17 @@ import (
 )
 
 var (
-	AsynqClient   *asynq.Client
 	scheduler     *asynq.Scheduler
 	schedulerOnce sync.Once
 )
 
 func init() {
-	AsynqClient = asynq.NewClient(task.RedisOpt)
+	// AsynqClient 已在 task 包中初始化
+}
+
+// GetAsynqClient 获取全局 AsynqClient
+func GetAsynqClient() *asynq.Client {
+	return task.AsynqClient
 }
 
 // StartScheduler 启动调度器

internal/task/utils.go

@@ -24,8 +24,12 @@ import (
 // RedisOpt asynq Redis 连接配置（兼容 Standalone/Sentinel/Cluster）
 var RedisOpt asynq.RedisConnOpt
 
+// AsynqClient asynq 客户端，用于任务入队
+var AsynqClient *asynq.Client
+
 func init() {
 	RedisOpt = NewRedisConnOpt()
+	AsynqClient = asynq.NewClient(RedisOpt)
 }
 
 // NewRedisConnOpt 根据配置返回对应的 asynq Redis 连接选项

internal/task/worker/middlewares.go

@@ -18,67 +18,15 @@ package worker
 
 import (
 	"context"
-	"time"
-
-	"github.com/linux-do/credit/internal/logger"
-	"go.opentelemetry.io/otel/trace"
 
 	"github.com/hibiken/asynq"
-	"github.com/linux-do/credit/internal/otel_trace"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
 )
 
-// taskLoggingMiddleware 记录任务日志中间件
+// taskLoggingMiddleware 任务处理中间件
+// 注意：OTel Span 创建、日志记录、TaskExecution 状态管理
+// 已由 task.ProcessTask 统一处理，此中间件保留用于未来扩展（如限流、监控等）
 func taskLoggingMiddleware(h asynq.Handler) asynq.Handler {
 	return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
-		// 初始化 Trace
-		ctx, span := otel_trace.Start(ctx, "TaskProcess_"+t.Type(), trace.WithSpanKind(trace.SpanKindConsumer))
-		defer span.End()
-
-		// 添加任务信息到 Span
-		span.SetAttributes(
-			attribute.String("task.type", t.Type()),
-			attribute.Int("task.payload_size", len(t.Payload())),
-			attribute.String("task.id", t.ResultWriter().TaskID()),
-		)
-
-		// 开始计时
-		start := time.Now()
-
-		// 处理任务
-		err := h.ProcessTask(ctx, t)
-
-		// 计算耗时
-		latency := time.Since(start)
-
-		if err != nil {
-			// 处理出错，记录错误日志
-			logger.ErrorF(
-				ctx,
-				"[TaskMiddleware] 任务处理失败 Type: %s\nStartTime: %s\nLatency: %d ms\nError: %v",
-				t.Type(),
-				start.Format(time.RFC3339),
-				latency.Milliseconds(),
-				err,
-			)
-
-			// 设置 Span 错误状态
-			span.SetStatus(codes.Error, err.Error())
-			span.RecordError(err)
-			return err
-		}
-
-		// 处理成功，记录成功日志
-		logger.InfoF(
-			ctx,
-			"[TaskMiddleware] 任务处理完成 Type: %s\nStartTime: %s\nEndTime: %s\nLatency: %d ms",
-			t.Type(),
-			start.Format(time.RFC3339),
-			time.Now().Format(time.RFC3339),
-			latency.Milliseconds(),
-		)
-
-		return nil
+		return h.ProcessTask(ctx, t)
 	})
 }

internal/task/worker/worker.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2025 linux.do
+Copyright 2025-2026 linux.do
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -25,6 +25,11 @@ import (
 	"github.com/linux-do/credit/internal/task"
 )
 
+func init() {
+	// 注册所有任务处理器
+	task.RegisterHandler(task.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{})
+}
+
 // StartWorker 启动任务处理服务器
 func StartWorker() error {
 	asynqServer := asynq.NewServer(
@@ -37,10 +42,13 @@ func StartWorker() error {
 		},
 	)
 
-	// 注册任务处理器
+	// 注册 Asynq 任务路由
 	mux := asynq.NewServeMux()
 	mux.Use(taskLoggingMiddleware)
-	mux.HandleFunc(task.CleanupUnusedUploadsTask, upload.HandleCleanupUnusedUploads)
+
+	// 统一使用 task.ProcessTask 处理所有任务类型
+	// 框架内部自动分发到对应的 TaskHandler 实现
+	mux.HandleFunc(task.CleanupUnusedUploadsTask, task.ProcessTask)
 
 	// 启动服务器
 	return asynqServer.Run(mux)