DEV-MERGE-LEGACY-BACKEND-2026/internal/task/task.go

// Package task 统一 DB 任务表 + 工作池（替代旧两套队列 t_queue + t_agent_queue，架构规格 §3.7）。
//
// 设计要点（与 ADR-015 单机双实例对齐）：
//   - 单状态机：pending → running → done / dead（双实例共享 t_task，靠乐观锁认领防重复消费）；
//   - 认领：UPDATE ... SET status='running', owner=?, claimed_at=? WHERE id=? AND status='pending'
//     （RowsAffected==1 才算认领成功——双实例并发认领同一任务时只有一个成功，另一个 RowsAffected==0 跳过）；
//   - 重试：失败按指数退避 next_run_at = now + 2^retry × 基数（1s→2s→4s→…→上限 1h），
//     retry 超过 TaskCfg.MaxRetry 置 dead（进告警 + 人工介入）；
//   - 幂等键：type + 业务唯一键（如 "ai_grading:{campus_id}:{homework_id}"）——
//     派发侧写入前查重，重投/重复请求不产生重复任务；
//   - **派发成功 ≠ 任务完成**（AI 批改/分析/导出/PDF）：前端轮询任务终态或业务表，
//     Agent 不可达只记日志不阻断主链路（fire-and-forget 语义，重活靠本表重试）。
//
// 典型使用：
//
//	// 派发（业务域，事务内或事务后）
//	t := task.New{Type: "ai_grading", CampusID: 3, Payload: payload, IdempotentKey: fmt.Sprintf("ai_grading:%d:%d", campusID, hwID)}
//	if err := repo.Enqueue(ctx, t); err != nil { ... } // 幂等键冲突时静默跳过（已存在等价任务）
//
//	// 工作池消费（cmd/api 启动的长驻 goroutine）
//	pool := task.NewPool(db, cfg.Task, redisClient, dispatcher)
//	go pool.Run(ctx) // ctx 取消时排空在途任务后退出（优雅停机，配合 GracePeriodSec）
package task

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"gorm.io/gorm"
)

// Status 任务状态机（t_task.status 列取值，单状态机——禁止新增中间态，
// 需要细分进度时在 payload 内表达，不改列枚举）。
const (
	StatusPending = "pending" // 已入队，待认领
	StatusRunning = "running" // 已被某实例认领，执行中
	StatusDone    = "done"    // 执行成功（终态）
	StatusDead    = "dead"    // 重试超限放弃（终态，进告警）
)

// Task 统一任务表 t_task 的领域模型（列与 DDL 一一对应，见 unified/02-data-model）。
//
// 字段：
//   - ID         int64：主键自增（仅内部排序用；对外暴露 idempotentKey，不暴露自增 ID）。
//   - Type       string：任务类型（"ai_grading"/"export_report"/"pdf_render"…；
//     新类型需在注册表注册 handler，未注册类型消费时直接置 dead + 告警）。
//   - CampusID   int64：任务归属校区（0=平台级任务；多租户作用域过滤用）。
//   - Payload    string：业务载荷 JSON（字符串直存，避免 GORM 序列化自定义类型；
//     消费侧反序列化失败 → 置 dead + 告警，不重试——载荷损坏重试也不会好）。
//   - Status     string：Status* 之一。
//   - IdempotentKey string：幂等键（type + 业务唯一键；UNIQUE 约束，防重）。
//   - Retry      int：已重试次数（每次失败 +1；超过 MaxRetry 置 dead）。
//   - NextRunAt  time.Time：下次可执行时刻（指数退避由 Fail 计算；pending 且 NextRunAt 未到不可被认领）。
//   - Owner      string：认领实例标识（"instance-a"/"instance-b"，ADR-015 双实例审计用；
//     实例崩溃后靠 claimed_at 超时回收，见 reclaimStale）。
//   - TraceID    string：贯穿 traceId（派发侧从请求 context 取；消费侧日志串联）。
//   - ClaimedAt  time.Time：认领时刻（stale 回收依据：running 且 now - claimed_at > 超时阈值 → 可被重新认领）。
//   - CreatedAt  time.Time：入队时刻。
type Task struct {
	ID            int64     `gorm:"column:id"`
	Type          string    `gorm:"column:type"`
	CampusID      int64     `gorm:"column:campus_id"`
	Payload       string    `gorm:"column:payload"`
	Status        string    `gorm:"column:status"`
	IdempotentKey string    `gorm:"column:idempotent_key"`
	Retry         int       `gorm:"column:retry"`
	NextRunAt     time.Time `gorm:"column:next_run_at"`
	Owner         string    `gorm:"column:owner"`
	TraceID       string    `gorm:"column:trace_id"`
	ClaimedAt     time.Time `gorm:"column:claimed_at"`
	CreatedAt     time.Time `gorm:"column:created_at"`
}

// New 待入队任务构造器（业务域派发侧使用；只填业务字段，状态/时间戳由 Enqueue 填充）。
type New struct {
	Type          string // 任务类型（必填，注册表内）
	CampusID      int64  // 校区（0=平台级）
	Payload       any    // 业务载荷（任意可 JSON 序列化结构）
	IdempotentKey string // 幂等键（必填；形如 "ai_grading:{campus_id}:{homework_id}"）
	TraceID       string // 贯穿 traceId（可空，空则消费侧重新生成）
}

// Repo 任务表仓储（GORM 实现；独立于业务域 Repository——task 域是横切域，
// 不继承 scope.BaseRepo：任务表本身平台级，作用域过滤在"派发侧"由业务域完成，
// 消费侧按 Type 路由到对应 handler，handler 内部自带作用域）。
type Repo struct {
	db *gorm.DB // 由 cmd/api 统一建立连接池后注入
}

// NewRepo 构造任务仓储。
//
// 入参：
//   - db *gorm.DB：数据库句柄（连接池由 infra/db 统一建立）。
//
// 出参：
//   - *Repo：就绪的仓储实例（db 为 nil 时各方法直接 panic——装配错误 fail-fast，单测覆盖）。
func NewRepo(db *gorm.DB) *Repo { return &Repo{db: db} }

// Enqueue 幂等入队（写 t_task，status=pending）。
//
// 入参：
//   - ctx context.Context：请求上下文（GORM WithContext 传递 traceId/取消信号）。
//   - n   New：待入队任务（Type/IdempotentKey 必填）。
//
// 出参：
//   - int64：新任务 ID（幂等命中已有任务时返回已有任务 ID，不报错——
//     "重复派发"是正常业务（用户连点两次提交），不是错误）。
//   - error：
//     - Type 或 IdempotentKey 为空 → 参数错误（调用方 bug）；
//     - Payload 序列化失败 → 包装 json 错误（载荷不可序列化属编码 bug，快速失败）；
//     - 唯一键冲突（并发双请求同幂等键）：GORM 返回 duplicate entry，本函数捕获后
//       回查已有任务 ID 返回（等价于成功，幂等语义）；
//     - 其它 DB 异常 → 包装 ErrDBError 语义（调用方决策：AI 类可降级重试，强一致类上报）。
//
// 样例：
//
//	id, err := repo.Enqueue(ctx, task.New{
//	    Type: "ai_grading", CampusID: 3,
//	    Payload:       GradingPayload{HomeworkID: 42},
//	    IdempotentKey: fmt.Sprintf("ai_grading:3:42"),
//	})
func (r *Repo) Enqueue(ctx context.Context, n New) (int64, error) {
	if n.Type == "" || n.IdempotentKey == "" {
		return 0, fmt.Errorf("task: Enqueue 要求 Type 与 IdempotentKey 非空")
	}
	payload, err := json.Marshal(n.Payload)
	if err != nil {
		return 0, fmt.Errorf("task: 载荷序列化失败(type=%s): %w", n.Type, err)
	}
	now := time.Now()
	t := &Task{
		Type:          n.Type,
		CampusID:      n.CampusID,
		Payload:       string(payload),
		Status:        StatusPending,
		IdempotentKey: n.IdempotentKey,
		NextRunAt:     now,
		TraceID:       n.TraceID,
		CreatedAt:     now,
	}
	res := r.db.WithContext(ctx).Create(t)
	if res.Error == nil {
		return t.ID, nil
	}
	// 幂等键唯一冲突 → 回查已有任务 ID（并发重复派发场景）
	if isDuplicateEntry(res.Error) {
		var existing Task
		if err := r.db.WithContext(ctx).
			Where("idempotent_key = ?", n.IdempotentKey).
			Take(&existing).Error; err == nil { // Take：存在性判断标准姿势（架构规格 §3.6）
			return existing.ID, nil
		}
	}
	return 0, fmt.Errorf("task: 入队失败(type=%s): %w", n.Type, res.Error)
}

// Claimable 拉取一批可认领任务（status=pending 且 next_run_at<=now；按创建序 FIFO）。
//
// 入参：
//   - ctx  context.Context：请求上下文。
//   - limit int：本批上限（TaskCfg.ClaimBatch，建议 10——防单实例饿死另一实例）。
//
// 出参：
//   - []Task：候选任务（可能为空 slice，非 nil）；**注意**：此处仅 SELECT，未加锁，
//     候选≠已认领——必须逐个经 Claim 乐观锁确认后才算本实例的任务。
//   - error：DB 异常时非 nil（调用方：本轮跳过 + 记 WARN，不退出工作池）。
func (r *Repo) Claimable(ctx context.Context, limit int) ([]Task, error) {
	if limit <= 0 {
		limit = 10
	}
	var list []Task
	err := r.db.WithContext(ctx).
		Where("status = ? AND next_run_at <= ?", StatusPending, time.Now()).
		Order("id ASC"). // FIFO：先入队先消费（公平性）
		Limit(limit).
		Find(&list).Error
	if err != nil {
		return nil, fmt.Errorf("task: 拉取可认领任务失败: %w", err)
	}
	return list, nil
}

// Claim 乐观锁认领单个任务（双实例防重复消费的核心）。
//
// 入参：
//   - ctx  context.Context：请求上下文。
//   - id   int64：候选任务 ID（来自 Claimable）。
//   - owner string：认领实例标识（"instance-a"/"instance-b"）。
//
// 出参：
//   - bool：true 认领成功（本实例独占该任务，开始执行）；
//     false 认领失败（已被另一实例抢先认领，或状态已非 pending——直接跳过，不是错误）。
//   - error：DB 异常时非 nil（调用方：本轮放弃该任务，下轮再试；不置 dead——
//     "认领失败"与"执行失败"语义不同，前者不该消耗重试次数）。
//
// SQL 语义（单条 UPDATE 原子化，无 SELECT 竞态）：
//
//	UPDATE t_task SET status='running', owner=?, claimed_at=NOW()
//	WHERE id=? AND status='pending'
//	RowsAffected==1 → 成功；==0 → 已被抢/状态已变
//
// 说明：不用 SELECT ... FOR UPDATE（PolarDB 行锁 + 长事务风险）——
// 条件 UPDATE 的 RowsAffected 判定是 InnoDB 下的标准乐观并发做法。
func (r *Repo) Claim(ctx context.Context, id int64, owner string) (bool, error) {
	res := r.db.WithContext(ctx).
		Model(&Task{}).
		Where("id = ? AND status = ?", id, StatusPending).
		Updates(map[string]interface{}{
			"status":     StatusRunning,
			"owner":      owner,
			"claimed_at": time.Now(),
		})
	if res.Error != nil {
		return false, fmt.Errorf("task: 认领任务 %d 失败: %w", id, res.Error)
	}
	return res.RowsAffected == 1, nil // 0 = 被抢或状态已变
}

// Done 标记任务成功（终态）。
//
// 入参：
//   - ctx context.Context：请求上下文。
//   - id  int64：任务 ID（调用方必须是该任务的 owner——本函数不做 owner 校验，
//     依赖"认领成功才执行"的调用纪律；审计靠 owner 列回溯）。
//
// 出参：
//   - error：仅 DB 异常时非 nil（调用方：重试标记操作；任务实际已成功，
//     标记失败最坏后果是任务"僵尸 running"——靠 reclaimStale 回收后重复执行，
//     handler 必须幂等（按 idempotentKey 查业务表终态直接返回成功））。
func (r *Repo) Done(ctx context.Context, id int64) error {
	res := r.db.WithContext(ctx).
		Model(&Task{}).
		Where("id = ? AND status = ?", id, StatusRunning).
		Update("status", StatusDone)
	if res.Error != nil {
		return fmt.Errorf("task: 标记成功(id=%d)失败: %w", id, res.Error)
	}
	return nil
}

// Fail 标记任务失败并计算下次重试（指数退避；超限置 dead）。
//
// 入参：
//   - ctx      context.Context：请求上下文。
//   - id       int64：任务 ID。
//   - maxRetry int：重试上限（TaskCfg.MaxRetry，建议 6：1s,2s,4s,8s,16s,32s 后 dead）。
//   - reason   string：失败原因摘要（≤256 字符，进日志/告警；**禁止**塞堆栈全文）。
//
// 出参：
//   - bool：true 任务仍存活（已安排 next_run_at 重试）；false 已置 dead（重试超限，告警+人工）。
//   - error：DB 异常时非 nil。
//
// 退避公式：next_run_at = now + min(2^retry 秒, 3600 秒)
// （retry 为本函数执行**前**的已重试次数：首次失败 retry=0→1s 后重试，第 6 次失败→dead）。
//
// 幂等性要求（调用方）：handler 失败路径若"业务已部分落库"，重试时必须走幂等键查终态，
// 否则退避重试会造成重复副作用（如重复批改写两份结果——业务表按 homework_id 唯一约束兜底）。
func (r *Repo) Fail(ctx context.Context, id int64, maxRetry int, reason string) (bool, error) {
	var t Task
	if err := r.db.WithContext(ctx).Where("id = ?", id).Take(&t).Error; err != nil {
		return false, fmt.Errorf("task: 读取任务 %d 失败: %w", id, err)
	}
	t.Retry++
	if t.Retry > maxRetry {
		res := r.db.WithContext(ctx).
			Model(&Task{}).
			Where("id = ? AND status = ?", id, StatusRunning).
			Update("status", StatusDead)
		if res.Error != nil {
			return false, fmt.Errorf("task: 置 dead(id=%d)失败: %w", id, res.Error)
		}
		return false, nil // dead：调用方记 ERROR 日志 + Prometheus 计数器 task_dead_total
	}
	backoff := time.Duration(1<<uint(t.Retry)) * time.Second // 2^retry 秒
	if backoff > time.Hour {
		backoff = time.Hour // 上限 1h（防长尾任务堆积时退避过久）
	}
	res := r.db.WithContext(ctx).
		Model(&Task{}).
		Where("id = ? AND status = ?", id, StatusRunning).
		Updates(map[string]interface{}{
			"status":      StatusPending,
			"retry":       t.Retry,
			"next_run_at": time.Now().Add(backoff),
		})
	if res.Error != nil {
		return false, fmt.Errorf("task: 安排重试(id=%d)失败: %w", id, res.Error)
	}
	return true, nil
}

// ReclaimStale 回收僵死任务（实例崩溃/卡死：running 且 claimed_at 超过超时阈值 → 重置 pending）。
//
// 入参：
//   - ctx       context.Context：请求上下文。
//   - staleAfter time.Duration：僵死判定阈值（建议 10min：Agent 长任务同步调用超时 60s ×
//     重试 2 + 余量；**必须大于** handler 最长执行时间，否则健康任务会被误回收重复执行）。
//
// 出参：
//   - int64：本次回收的任务数（0=无僵死，正常）。
//   - error：DB 异常时非 nil（调用方：下一周期再试；本函数由工作池侧周期调用，非请求链路）。
//
// 语义：回收后任务回到 pending，可能被另一实例重新认领——handler 幂等是最后一道防线。
//
// 样例（工作池每 60s 调一次）：
//
//	n, _ := repo.ReclaimStale(ctx, 10*time.Minute)
//	if n > 0 { log.Warn("task: 回收僵死任务", zap.Int64("count", n)) }
func (r *Repo) ReclaimStale(ctx context.Context, staleAfter time.Duration) (int64, error) {
	res := r.db.WithContext(ctx).
		Model(&Task{}).
		Where("status = ? AND claimed_at < ?", StatusRunning, time.Now().Add(-staleAfter)).
		Updates(map[string]interface{}{
			"status": StatusPending,
			"owner":  "",
		})
	if res.Error != nil {
		return 0, fmt.Errorf("task: 回收僵死任务失败: %w", res.Error)
	}
	return res.RowsAffected, nil
}

// isDuplicateEntry 判定 MySQL 唯一键冲突错误（匹配 "Duplicate entry" 文案，
// 不依赖具体驱动错误类型——PolarDB 兼容 MySQL 协议，驱动错误包装层级可能不同，
// 文案匹配最稳）。
//
// 入参：err error：GORM 返回的原始错误。
// 出参：bool：true 表示唯一键冲突（幂等命中）。
func isDuplicateEntry(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "Duplicate entry") || strings.Contains(msg, "Error 1062")
}