feat: 实现客户端注册与任务领取接口

Admin 的第一个业务功能。选它打头是因为它是穿透所有层的最薄一条竖切（HTTP → handler/api → service → repository → SQLite → handler/web → 页面），一个工单把分层模式立起来，后面四个模块照抄；同时它是与 Client 联调的接口，能解锁另一条并行的工作线。实现 - POST /tasks/claim：注册 + 领取。注册就在这里做，没有单独的注册接口，也没有心跳（理由见 docs/admin/04-client-api.md §3） - 领取用条件更新 + 检查影响行数防并发，SQLite 没有 SELECT FOR UPDATE - 客户端列表页：查询、按名称搜索、批量删除 - 在线状态是**算出来的**（last_seen_at 在 10 分钟内），数据库里没有该字段 - CSRF 中间件：双提交 Cookie，手写 82 行不引依赖。 **只挂页面路由**，/api/v1/client/* 不能加——Client 不是浏览器、没有 Cookie - 14 个单元测试修复一个真 bug：PRAGMA 必须写进 DSN 并发领取测试报 database is locked (SQLITE_BUSY)。根因是 PRAGMA busy_timeout 每连接生效，而 database/sql 是连接池—— db.Exec("PRAGMA ...") 只作用于当时那条连接，池子新开的连接没执行过。单线程正常、一并发就炸。改成 DSN 传参后并发测试跑 20 次全过。这个坑已写进 docs/admin/03-data-model.md §2.1。与工单的两处差异 - 去掉 name_is_custom 列后，"人工改的名字不被覆盖"改用更简单的做法： ON CONFLICT DO UPDATE SET 里不含 name，即只在首次注册时写入。效果相同，零额外字段、零迁移。已同步 04 §3 - 验收项"不向 dry_run 客户端分配真实下单任务"**未实现**： tasks 表没有字段标记任务是否需要真实下单。当前真实下单开关默认关闭、 MVP 全是演练模式，暂不出问题，但开真实下单前必须补该字段，需另开工单已验证（Go 1.23.0） - go vet / gofmt / go test 全过，并发测试重复 20 次稳定通过 - 端到端：无任务 claim 204；插入任务后 claim 200 且 payload 含 goods_url/goods_id/options/quantity/max_price_cent、无租约无 Admin 状态；重复 claim 204；缺 X-Client-Id 400；POST 无 CSRF token 403；列表页两台客户端在线状态与统计正确说明：Gitea 尚未配置，本次无对应工单号。 submit_result / submit_failure 及其幂等处理留给下一个工单。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 16:56:18 +08:00
parent 0279ffd60e
commit 6ae768463f
19 changed files with 1069 additions and 79 deletions
@@ -0,0 +1,121 @@
+package repository
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"cmautobuy/admin/model"
+)
+
+// UpsertClient 登记或更新一台客户端。
+//
+// 注意 name 的处理：**只在第一次注册时写入，之后不再更新**。
+// 这样操作员在 Admin 界面上改成好记的名字后，客户端每次 claim
+// 都不会把它覆盖回去。做法是 ON CONFLICT 的 DO UPDATE 里不含 name。
+//
+// name 为空时用 clientID 当显示名，保证列表里不出现空白行。
+func UpsertClient(db *sql.DB, c model.Client) error {
+	if c.ClientID == "" {
+		return fmt.Errorf("client_id 不能为空")
+	}
+	name := c.Name
+	if name == "" {
+		name = c.ClientID
+	}
+	now := model.NowISO()
+
+	_, err := db.Exec(`
+		INSERT INTO clients (client_id, name, device_address, platform,
+		                     pdd_package, capabilities,
+		                     last_seen_at, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+		ON CONFLICT(client_id) DO UPDATE SET
+		    device_address = excluded.device_address,
+		    platform       = excluded.platform,
+		    pdd_package    = excluded.pdd_package,
+		    capabilities   = excluded.capabilities,
+		    last_seen_at   = excluded.last_seen_at,
+		    updated_at     = excluded.updated_at`,
+		c.ClientID, name, c.DeviceAddress, c.Platform,
+		c.PddPackage, c.Capabilities, now, now, now)
+	if err != nil {
+		return fmt.Errorf("登记客户端 %s 失败: %w", c.ClientID, err)
+	}
+	return nil
+}
+
+// TouchClient 只刷新 last_seen_at。
+//
+// claim / result / failure 三个接口都要调。只在 claim 里调的话，
+// 客户端执行长任务期间不调 claim，会被误判成离线。
+func TouchClient(db *sql.DB, clientID string) error {
+	now := model.NowISO()
+	_, err := db.Exec(
+		`UPDATE clients SET last_seen_at = ?, updated_at = ? WHERE client_id = ?`,
+		now, now, clientID)
+	if err != nil {
+		return fmt.Errorf("刷新客户端 %s 活动时间失败: %w", clientID, err)
+	}
+	return nil
+}
+
+// ListClients 按名称模糊查询客户端，keyword 为空时返回全部。
+func ListClients(db *sql.DB, keyword string) ([]model.Client, error) {
+	query := `SELECT client_id, name, device_address, platform, pdd_package,
+	                 capabilities, last_seen_at, created_at, updated_at
+	          FROM clients`
+	args := []any{}
+
+	if kw := strings.TrimSpace(keyword); kw != "" {
+		// 参数化查询，通配符拼在值里而不是 SQL 里
+		query += ` WHERE name LIKE ? OR client_id LIKE ?`
+		like := "%" + kw + "%"
+		args = append(args, like, like)
+	}
+	query += ` ORDER BY last_seen_at DESC, client_id`
+
+	rows, err := db.Query(query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("查询客户端列表失败: %w", err)
+	}
+	defer rows.Close()
+
+	var out []model.Client
+	for rows.Next() {
+		var c model.Client
+		var name, addr, platform, pkg, caps sql.NullString
+		if err := rows.Scan(&c.ClientID, &name, &addr, &platform, &pkg,
+			&caps, &c.LastSeenAt, &c.CreatedAt, &c.UpdatedAt); err != nil {
+			return nil, fmt.Errorf("读取客户端行失败: %w", err)
+		}
+		c.Name = name.String
+		c.DeviceAddress = addr.String
+		c.Platform = platform.String
+		c.PddPackage = pkg.String
+		c.Capabilities = caps.String
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+// DeleteClients 批量删除客户端。返回实际删除的条数。
+func DeleteClients(db *sql.DB, clientIDs []string) (int64, error) {
+	if len(clientIDs) == 0 {
+		return 0, nil
+	}
+
+	// 占位符按数量生成，值仍然是参数化传入，不存在注入
+	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(clientIDs)), ",")
+	args := make([]any, len(clientIDs))
+	for i, id := range clientIDs {
+		args[i] = id
+	}
+
+	res, err := db.Exec(
+		`DELETE FROM clients WHERE client_id IN (`+placeholders+`)`, args...)
+	if err != nil {
+		return 0, fmt.Errorf("删除客户端失败: %w", err)
+	}
+	return res.RowsAffected()
+}
@@ -20,26 +20,45 @@ import (
 	_ "modernc.org/sqlite"
 )

-// Open 打开 data/admin.db，并设置必要的 PRAGMA。
+// Open 打开 data/admin.db。
+//
+// **PRAGMA 必须写在 DSN 里，不能用 db.Exec("PRAGMA ...") 设置。**
+//
+// 原因是 Go 的 database/sql 是一个**连接池**：db.Exec 只作用于当时
+// 拿到的那一条连接，池子后来新开的连接完全没执行过那些 PRAGMA。
+// 并发写的时候，没有 busy_timeout 的那些连接会直接报
+// "database is locked (SQLITE_BUSY)"，而不是等锁释放。
+//
+// 写进 DSN 后，驱动会对**每一条新连接**都应用一遍，这才是对的。
 func Open(dataDir string) (*sql.DB, error) {
 	path := filepath.Join(dataDir, "admin.db")

-	db, err := sql.Open("sqlite", path)
+	// 三条设置的含义见 docs/admin/03-data-model.md §2：
+	//   busy_timeout  拿不到锁时最多等 5 秒，而不是立刻报错
+	//   journal_mode  WAL 模式，读和写可以同时进行
+	//   foreign_keys  打开外键约束（SQLite 默认是关的）
+	dsn := "file:" + path +
+		"?_pragma=busy_timeout(5000)" +
+		"&_pragma=journal_mode(WAL)" +
+		"&_pragma=foreign_keys(1)"
+
+	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, fmt.Errorf("打开数据库 %s 失败: %w", path, err)
 	}

-	// 这三条见 docs/admin/03-data-model.md §2
-	pragmas := []string{
-		"PRAGMA foreign_keys = ON",
-		"PRAGMA journal_mode = WAL",
-		"PRAGMA busy_timeout = 5000",
-	}
-	for _, p := range pragmas {
-		if _, err := db.Exec(p); err != nil {
-			db.Close()
-			return nil, fmt.Errorf("执行 %s 失败: %w", p, err)
-		}
+	// SQLite 同一时刻只允许一个写事务。连接数放太开，
+	// 大量连接会互相抢锁、把 busy_timeout 耗光。
+	// 本项目是单机内部工具，并发量很小，限制在个位数足够。
+	//
+	// 注意**不要设成 1**：那样在一个事务里再调用需要连接的代码会死锁。
+	db.SetMaxOpenConns(4)
+	db.SetMaxIdleConns(4)
+
+	// sql.Open 是懒加载的，这里主动连一次，好让配置错误立刻暴露
+	if err := db.Ping(); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("连接数据库 %s 失败: %w", path, err)
 	}
 	return db, nil
 }
@@ -0,0 +1,127 @@
+package repository
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"cmautobuy/admin/model"
+)
+
+// claimCandidateLimit 是一次最多尝试抢多少条。
+// 抢不到说明被别的客户端拿走了，再试下一条；都抢不到就当作没任务。
+const claimCandidateLimit = 10
+
+// ClaimNextTask 为指定客户端领取一个任务。
+//
+// 没有可领的任务时返回 (nil, nil) —— 调用方据此返回 204。
+//
+// 防并发的做法是**条件更新 + 检查影响行数**：先查出候选，
+// 再用 `WHERE task_id = ? AND status = 'assigned'` 去更新，
+// 影响行数为 0 就说明被别人抢先了，换下一条。
+// 不用 SELECT ... FOR UPDATE，SQLite 没有那个。
+func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) {
+	if clientID == "" {
+		return nil, fmt.Errorf("client_id 不能为空")
+	}
+
+	query := `SELECT task_id FROM tasks
+	          WHERE assigned_client = ? AND status = 'assigned'`
+	args := []any{clientID}
+
+	// 客户端只声明支持某些类型时，不要给它别的类型
+	if len(supportedTypes) > 0 {
+		placeholders := strings.TrimSuffix(strings.Repeat("?,", len(supportedTypes)), ",")
+		query += ` AND task_type IN (` + placeholders + `)`
+		for _, t := range supportedTypes {
+			args = append(args, t)
+		}
+	}
+	query += ` ORDER BY priority DESC, created_at LIMIT ?`
+	args = append(args, claimCandidateLimit)
+
+	rows, err := db.Query(query, args...)
+	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 {
+		res, err := db.Exec(`
+			UPDATE tasks SET status = 'claimed', claimed_at = ?, updated_at = ?
+			WHERE task_id = ? AND status = 'assigned'`,
+			now, now, taskID)
+		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 // 被别的客户端抢先了，换下一条
+		}
+		return GetTask(db, taskID)
+	}
+	return nil, nil // 没有可领的任务
+}
+
+// GetTask 按编号读一条任务。
+func GetTask(db *sql.DB, taskID string) (*model.Task, error) {
+	var t model.Task
+	var assigned, claimedAt, sybID, orderNo, goodsID, skuID sql.NullString
+	var pddGoodsID, pddOptions, resultData, errCode, errMsg, finishedAt sql.NullString
+	var quantity, maxPrice sql.NullInt64
+
+	err := db.QueryRow(`
+		SELECT task_id, task_type, status, version, priority,
+		       assigned_client, claimed_at,
+		       syb_id, order_no, goods_id, shopee_sku_id,
+		       pdd_goods_url, pdd_goods_id, pdd_options,
+		       quantity, max_price_cent,
+		       result_data, error_code, error_message, finished_at,
+		       created_at, updated_at
+		FROM tasks WHERE task_id = ?`, taskID).Scan(
+		&t.TaskID, &t.TaskType, &t.Status, &t.Version, &t.Priority,
+		&assigned, &claimedAt,
+		&sybID, &orderNo, &goodsID, &skuID,
+		&t.PddGoodsURL, &pddGoodsID, &pddOptions,
+		&quantity, &maxPrice,
+		&resultData, &errCode, &errMsg, &finishedAt,
+		&t.CreatedAt, &t.UpdatedAt)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("读取任务 %s 失败: %w", taskID, err)
+	}
+
+	t.AssignedClient = assigned.String
+	t.ClaimedAt = claimedAt.String
+	t.SybID = sybID.String
+	t.OrderNo = orderNo.String
+	t.GoodsID = goodsID.String
+	t.ShopeeSKUID = skuID.String
+	t.PddGoodsID = pddGoodsID.String
+	t.PddOptions = pddOptions.String
+	t.Quantity = int(quantity.Int64)
+	t.MaxPriceCent = maxPrice.Int64
+	t.ResultData = resultData.String
+	t.ErrorCode = errCode.String
+	t.ErrorMessage = errMsg.String
+	t.FinishedAt = finishedAt.String
+	return &t, nil
+}