feat: 迁移 Admin Repository 到 MySQL 8 (#79)

admin/repository/task.go

@@ -1,17 +1,15 @@
 package repository
 
 import (
+	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"strings"
 
 	"cmautobuy/admin/model"
 )
 
-// claimCandidateLimit 是一次最多尝试抢多少条。
-// 抢不到说明被别的客户端拿走了，再试下一条；都抢不到就当作没任务。
-const claimCandidateLimit = 10
-
 // ClaimNextTask 为指定客户端领取一个任务。
 //
 // 没有可领的任务时返回 (nil, nil) —— 调用方据此返回 204。
@@ -30,13 +28,18 @@ const claimCandidateLimit = 10
 //
 // # 防并发
 //
-// 做法是**条件更新 + 检查影响行数**：先查出候选，再带着状态条件去更新，
-// 影响行数为 0 就说明被别人抢先了，换下一条。
-// 不用 SELECT ... FOR UPDATE，SQLite 没有那个。
+// InnoDB 事务用 FOR UPDATE SKIP LOCKED 锁住一条候选任务。领取状态和
+// task_claims 历史在同一个事务提交，避免只改了状态却没留下领取凭据。
 func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) {
 	if clientID == "" {
 		return nil, fmt.Errorf("client_id 不能为空")
 	}
+	ctx := context.Background()
+	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return nil, fmt.Errorf("开始领取任务事务失败: %w", err)
+	}
+	defer tx.Rollback()
 
 	query := `SELECT task_id FROM tasks
 	          WHERE (   (assigned_client = ?    AND status = 'assigned')
@@ -52,57 +55,45 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model
 		}
 	}
 	// (assigned_client IS NULL) 为 0/1，0 排前面 —— 指定给本机的优先于无主的
-	query += ` ORDER BY (assigned_client IS NULL), priority DESC, created_at LIMIT ?`
-	args = append(args, claimCandidateLimit)
-
-	rows, err := db.Query(query, args...)
-	if err != nil {
+	query += ` ORDER BY (assigned_client IS NULL), priority DESC, created_at
+	           LIMIT 1 FOR UPDATE SKIP LOCKED`
+	var taskID string
+	if err := tx.QueryRowContext(ctx, query, args...).Scan(&taskID); errors.Is(err, sql.ErrNoRows) {
+		return nil, nil
+	} else if err != nil {
 		return nil, fmt.Errorf("查询可领任务失败: %w", err)
 	}
-	var candidates []string
-	for rows.Next() {
-		var id string
-		if err := rows.Scan(&id); err != nil {
-			rows.Close()
-			return nil, fmt.Errorf("读取候选任务失败: %w", err)
-		}
-		candidates = append(candidates, id)
-	}
-	rows.Close()
-	if err := rows.Err(); err != nil {
-		return nil, err
-	}
 
 	now := model.NowISO()
-	for _, taskID := range candidates {
-		// 两种情况合成一条语句：对"指定给我的"那种，写 assigned_client
-		// 是写同一个值，无副作用；对无主的，这一步就是"谁领到就标记谁"。
-		res, err := db.Exec(`
+	// 两种情况合成一条语句：对"指定给我的"那种，写 assigned_client
+	// 是写同一个值，无副作用；对无主的，这一步就是"谁领到就标记谁"。
+	res, err := tx.ExecContext(ctx, `
 			UPDATE tasks
 			   SET status = 'claimed', assigned_client = ?,
 			       claimed_at = ?, updated_at = ?
 			 WHERE task_id = ?
 			   AND (   (status = 'assigned' AND assigned_client = ?)
 			        OR (status = 'pending'  AND assigned_client IS NULL) )`,
-			clientID, now, now, taskID, clientID)
-		if err != nil {
-			return nil, fmt.Errorf("领取任务 %s 失败: %w", taskID, err)
-		}
-		n, err := res.RowsAffected()
-		if err != nil {
-			return nil, err
-		}
-		if n == 0 {
-			continue // 被别的客户端抢先了，换下一条
-		}
-		// 记一笔领取历史。提交结果时要靠它判断这台客户端有没有领过——
-		// 任务重派后 assigned_client 会变，只看它就查不出来了。
-		if err := RecordClaim(db, taskID, clientID, now); err != nil {
-			return nil, err
-		}
-		return GetTask(db, taskID)
+		clientID, now, now, taskID, clientID)
+	if err != nil {
+		return nil, fmt.Errorf("领取任务 %s 失败: %w", taskID, err)
 	}
-	return nil, nil // 没有可领的任务
+	n, err := res.RowsAffected()
+	if err != nil {
+		return nil, fmt.Errorf("确认领取任务 %s 结果失败: %w", taskID, err)
+	}
+	if n != 1 {
+		return nil, fmt.Errorf("领取任务 %s 时状态异常", taskID)
+	}
+	// 记一笔领取历史。提交结果时要靠它判断这台客户端有没有领过——
+	// 任务重派后 assigned_client 会变，只看它就查不出来了。
+	if err := RecordClaim(tx, taskID, clientID, now); err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, fmt.Errorf("提交领取任务事务失败: %w", err)
+	}
+	return GetTask(db, taskID)
 }
 
 // GetTask 按编号读一条任务。
@@ -159,7 +150,7 @@ func RecordClaim(q Execer, taskID, clientID, claimedAt string) error {
 	_, err := q.Exec(`
 		INSERT INTO task_claims (task_id, client_id, claimed_at)
 		VALUES (?, ?, ?)
-		ON CONFLICT(task_id, client_id) DO UPDATE SET claimed_at = excluded.claimed_at`,
+		ON DUPLICATE KEY UPDATE claimed_at = VALUES(claimed_at)`,
 		taskID, clientID, claimedAt)
 	if err != nil {
 		return fmt.Errorf("记录领取历史失败 task=%s client=%s: %w", taskID, clientID, err)
@@ -312,7 +303,7 @@ func taskFilterClause(filter TaskFilter) (string, []any) {
 	if kw := strings.TrimSpace(filter.Keyword); kw != "" {
 		pattern := "%" + escapeLike(kw) + "%"
 		clauses = append(clauses,
-			"(t.task_id LIKE ? ESCAPE '\\' OR t.order_no LIKE ? ESCAPE '\\' OR t.pdd_goods_id LIKE ? ESCAPE '\\')")
+			"(t.task_id LIKE ? ESCAPE '!' OR t.order_no LIKE ? ESCAPE '!' OR t.pdd_goods_id LIKE ? ESCAPE '!')")
 		args = append(args, pattern, pattern, pattern)
 	}