feat: 安全回写档口入库码并支持核验恢复 (#234)

2026-08-15 09:10:10 +08:00
parent 15ebda23e3
commit 2778eed29f
11 changed files with 782 additions and 4 deletions
@@ -274,6 +274,102 @@ func scanInnerCodeRecord(scanner innerCodeRowScanner) (model.InnerCodeRecord, er
 	return row, nil
 }

+// ClaimInnerCodeForApply 原子领取一条 ready 记录。返回 claimed=false 表示状态已变化。
+func ClaimInnerCodeForApply(db *sql.DB, id int64, actorUserID, now string) (*model.InnerCodeRecord, bool, error) {
+	tx, err := db.Begin()
+	if err != nil {
+		return nil, false, fmt.Errorf("开始领取档口入库码事务失败: %w", err)
+	}
+	defer tx.Rollback()
+	record, err := scanInnerCodeRecord(tx.QueryRow(`SELECT `+innerCodeListColumns+
+		` FROM syb_inner_code_records WHERE id=? FOR UPDATE`, id))
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, false, nil
+	}
+	if err != nil {
+		return nil, false, err
+	}
+	if record.Status != model.InnerCodeReady {
+		return &record, false, nil
+	}
+	result, err := tx.Exec(`UPDATE syb_inner_code_records
+		SET status='applying',applied_by_user_id=?,apply_started_at=?,updated_at=?
+		WHERE id=? AND status='ready'`, actorUserID, now, now, id)
+	if err != nil {
+		return nil, false, fmt.Errorf("领取档口入库码记录失败: %w", err)
+	}
+	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
+		return &record, false, nil
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, false, fmt.Errorf("提交档口入库码领取失败: %w", err)
+	}
+	record.Status = model.InnerCodeApplying
+	record.AppliedByUserID = actorUserID
+	record.ApplyStartedAt = now
+	record.UpdatedAt = now
+	return &record, true, nil
+}
+
+// FinishInnerCodeApply 保存一条已领取记录的最终结果。
+func FinishInnerCodeApply(q Execer, id int64, status model.InnerCodeStatus, message, remoteCode, finishedAt string) error {
+	result, err := q.Exec(`UPDATE syb_inner_code_records
+		SET status=?,result_message=?,remote_inner_code=?,
+		    applied_at=CASE WHEN ? IN ('updated','already_filled') THEN ? ELSE applied_at END,
+		    updated_at=?
+		WHERE id=? AND status='applying'`, status, message, nullableString(remoteCode), status, finishedAt, finishedAt, id)
+	if err != nil {
+		return fmt.Errorf("保存档口入库码回写结果失败: %w", err)
+	}
+	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
+		return fmt.Errorf("档口入库码记录 %d 已不在回写中，拒绝覆盖结果", id)
+	}
+	return nil
+}
+
+// GetInnerCodeForRecheck 读取一条需核对记录。
+func GetInnerCodeForRecheck(q Execer, id int64) (*model.InnerCodeRecord, error) {
+	record, err := scanInnerCodeRecord(q.QueryRow(`SELECT `+innerCodeListColumns+
+		` FROM syb_inner_code_records WHERE id=?`, id))
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &record, nil
+}
+
+// SaveInnerCodeRecheck 保存只读重新核对的远端结果，不执行状态领取或写入。
+func SaveInnerCodeRecheck(q Execer, id int64, status model.InnerCodeStatus, message, remoteCode, checkedAt string) error {
+	result, err := q.Exec(`UPDATE syb_inner_code_records
+		SET status=?,result_message=?,remote_inner_code=?,
+		    applied_at=CASE WHEN ?='updated' THEN ? ELSE applied_at END,updated_at=?
+		WHERE id=? AND status='needs_check'`, status, message, nullableString(remoteCode), status, checkedAt, checkedAt, id)
+	if err != nil {
+		return fmt.Errorf("保存档口入库码核对结果失败: %w", err)
+	}
+	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
+		return fmt.Errorf("档口入库码记录 %d 已不需要核对", id)
+	}
+	return nil
+}
+
+// InterruptApplyingInnerCodes 在启动时把未知结果的 applying 收敛为 needs_check。
+func InterruptApplyingInnerCodes(q Execer, interruptedAt string) (int, error) {
+	result, err := q.Exec(`UPDATE syb_inner_code_records
+		SET status='needs_check',result_message='Admin 在回写完成前退出，请重新核对远端结果；系统不会自动重写',
+		    updated_at=? WHERE status='applying'`, interruptedAt)
+	if err != nil {
+		return 0, fmt.Errorf("恢复中断的档口入库码回写失败: %w", err)
+	}
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("读取中断档口入库码数量失败: %w", err)
+	}
+	return int(affected), nil
+}
+
 func nullablePositiveInt(value int) any {
 	if value <= 0 {
 		return nil
@@ -0,0 +1,40 @@
+package repository
+
+import (
+	"database/sql"
+	"testing"
+
+	_ "modernc.org/sqlite"
+)
+
+func TestInterruptApplyingInnerCodes_只收敛回写中记录(t *testing.T) {
+	db, err := sql.Open("sqlite", "file:inner_code_interrupt?mode=memory&cache=shared")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE syb_inner_code_records (
+		id INTEGER PRIMARY KEY,status TEXT NOT NULL,result_message TEXT,updated_at TEXT NOT NULL
+	)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`INSERT INTO syb_inner_code_records(id,status,updated_at) VALUES
+		(1,'applying','old'),(2,'ready','old')`); err != nil {
+		t.Fatal(err)
+	}
+	count, err := InterruptApplyingInnerCodes(db, "2026-08-15T01:00:00Z")
+	if err != nil || count != 1 {
+		t.Fatalf("count=%d err=%v", count, err)
+	}
+	var status, message, updatedAt string
+	if err := db.QueryRow(`SELECT status,result_message,updated_at FROM syb_inner_code_records WHERE id=1`).
+		Scan(&status, &message, &updatedAt); err != nil {
+		t.Fatal(err)
+	}
+	if status != "needs_check" || message == "" || updatedAt != "2026-08-15T01:00:00Z" {
+		t.Fatalf("status=%s message=%q updated=%s", status, message, updatedAt)
+	}
+	if err := db.QueryRow(`SELECT status FROM syb_inner_code_records WHERE id=2`).Scan(&status); err != nil || status != "ready" {
+		t.Fatalf("ready 记录不应变化 status=%s err=%v", status, err)
+	}
+}
@@ -1083,6 +1083,40 @@ func prepareMySQLV8(t *testing.T, db *sql.DB) {
 	mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES(8,'2026-08-11T00:00:00Z')`)
 }

+func TestMySQLMigrate_V22升级V23且唯一约束生效(t *testing.T) {
+	db := openMySQLMigrationTestDB(t)
+	defer db.Close()
+	cleanMySQLTestSchema(t, db)
+	defer cleanMySQLTestSchema(t, db)
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatal(err)
+	}
+	mustExec(t, db, `SET FOREIGN_KEY_CHECKS=0`)
+	mustExec(t, db, `DROP TABLE syb_inner_code_records`)
+	mustExec(t, db, `DELETE FROM schema_migrations WHERE version=23`)
+	mustExec(t, db, `SET FOREIGN_KEY_CHECKS=1`)
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatalf("v22 升级 v23 失败: %v", err)
+	}
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatalf("v23 重复迁移失败: %v", err)
+	}
+	if err := checkMySQLV23Shape(db); err != nil {
+		t.Fatal(err)
+	}
+	now := "2026-08-15T00:00:00Z"
+	mustExec(t, db, `INSERT INTO users(user_id,username,password_hash,role,status,password_changed_at,created_at,updated_at)
+		VALUES('inner-user','inner-user','hash','purchaser','active',?,?,?)`, now, now, now)
+	mustExec(t, db, `INSERT INTO syb_inner_code_records
+		(business_date,source_row,order_number,stall,spec_raw,spec_key,inner_code,status,created_by_user_id,created_at,updated_at)
+		VALUES('2026-08-15',2,'ORDER-1','A#1','黑色,M','黑色,M','DK-1','pending','inner-user',?,?)`, now, now)
+	if _, err := db.Exec(`INSERT INTO syb_inner_code_records
+		(business_date,source_row,order_number,stall,spec_raw,spec_key,inner_code,status,created_by_user_id,created_at,updated_at)
+		VALUES('2026-08-15',3,'ORDER-2','B#2','白色,L','白色,L','DK-1','pending','inner-user',?,?)`, now, now); err == nil {
+		t.Fatal("同一业务日期的 inner_code 唯一约束必须拒绝冲突")
+	}
+}
+
 func mustExec(t *testing.T, db *sql.DB, query string, args ...any) {
 	t.Helper()
 	if _, err := db.Exec(query, args...); err != nil {