feat: 实现 SQLite 到 MySQL 单向迁移 (#80)
This commit is contained in:
@@ -0,0 +1,75 @@
|
|||||||
|
// migrate-sqlite-to-mysql 把历史 SQLite v8 数据一次性导入空的 MySQL 8 数据库。
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"flag"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
|
||||||
|
"cmautobuy/admin/config"
|
||||||
|
"cmautobuy/admin/repository"
|
||||||
|
)
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
sourcePath := flag.String("source", "", "历史 admin.db 路径(必填,只读打开)")
|
||||||
|
dryRun := flag.Bool("dry-run", false, "只检查源库和空目标库,不写入")
|
||||||
|
execute := flag.Bool("execute", false, "执行一次性迁移")
|
||||||
|
verifyOnly := flag.Bool("verify-only", false, "只核对已迁移的 MySQL 与 SQLite")
|
||||||
|
flag.Parse()
|
||||||
|
|
||||||
|
if *sourcePath == "" {
|
||||||
|
log.Fatal("必须提供 --source 指向历史 admin.db")
|
||||||
|
}
|
||||||
|
modeCount := 0
|
||||||
|
for _, enabled := range []bool{*dryRun, *execute, *verifyOnly} {
|
||||||
|
if enabled {
|
||||||
|
modeCount++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if modeCount != 1 {
|
||||||
|
log.Fatal("必须且只能选择 --dry-run、--execute、--verify-only 其中一个")
|
||||||
|
}
|
||||||
|
|
||||||
|
source, err := repository.OpenLegacySQLiteReadOnly(*sourcePath)
|
||||||
|
if err != nil {
|
||||||
|
log.Fatal(err)
|
||||||
|
}
|
||||||
|
defer source.Close()
|
||||||
|
|
||||||
|
databaseConfig, err := config.LoadDatabaseFromEnv()
|
||||||
|
if err != nil {
|
||||||
|
log.Fatalf("读取 MySQL 配置失败: %v", err)
|
||||||
|
}
|
||||||
|
target, err := repository.OpenMySQL(databaseConfig)
|
||||||
|
if err != nil {
|
||||||
|
log.Fatal(err)
|
||||||
|
}
|
||||||
|
defer target.Close()
|
||||||
|
if err := repository.MigrateMySQL(target); err != nil {
|
||||||
|
log.Fatalf("准备 MySQL schema 失败: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var summary *repository.SQLiteMigrationSummary
|
||||||
|
switch {
|
||||||
|
case *verifyOnly:
|
||||||
|
summary, err = repository.VerifySQLiteToMySQL(source, target)
|
||||||
|
case *dryRun:
|
||||||
|
summary, err = repository.MigrateSQLiteToMySQL(source, target, true)
|
||||||
|
default:
|
||||||
|
summary, err = repository.MigrateSQLiteToMySQL(source, target, false)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
log.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
mode := "迁移完成并核对通过"
|
||||||
|
if *dryRun {
|
||||||
|
mode = "演练检查通过,未写入数据"
|
||||||
|
} else if *verifyOnly {
|
||||||
|
mode = "迁移数据核对通过,未写入数据"
|
||||||
|
}
|
||||||
|
fmt.Println(mode)
|
||||||
|
for _, table := range repository.SortedMigrationSummary(summary) {
|
||||||
|
fmt.Printf("%-28s %d\n", table.Name, table.Count)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,4 +1,5 @@
|
|||||||
// Package repository 封装 SQLite 读写。
|
// Package repository 封装数据库读写。本文件只保留历史 SQLite v8 schema 与迁移,
|
||||||
|
// 供一次性 SQLite→MySQL 工具和历史回归测试使用,不进入生产 Admin 启动路径。
|
||||||
//
|
//
|
||||||
// 改动本文件前必读 admin/AGENTS.md。三条硬规则:
|
// 改动本文件前必读 admin/AGENTS.md。三条硬规则:
|
||||||
// - 只有本包能写 SQL,handler 和 service 都不许拼 SQL;
|
// - 只有本包能写 SQL,handler 和 service 都不许拼 SQL;
|
||||||
|
|||||||
@@ -0,0 +1,354 @@
|
|||||||
|
package repository
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"crypto/sha256"
|
||||||
|
"database/sql"
|
||||||
|
"encoding/binary"
|
||||||
|
"fmt"
|
||||||
|
"path/filepath"
|
||||||
|
"sort"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// SQLiteMigrationSummary 只包含表名和数量,可以安全写入终端或工单。
|
||||||
|
type SQLiteMigrationSummary struct {
|
||||||
|
DryRun bool
|
||||||
|
Tables []SQLiteMigrationTable
|
||||||
|
}
|
||||||
|
|
||||||
|
type SQLiteMigrationTable struct {
|
||||||
|
Name string
|
||||||
|
Count int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// SQLiteMigrationError 隐藏可能带业务主键的数据库原始错误。
|
||||||
|
// Cause 只供 errors.Is/As 和测试使用,不应直接输出。
|
||||||
|
type SQLiteMigrationError struct {
|
||||||
|
Stage string
|
||||||
|
Table string
|
||||||
|
Cause error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *SQLiteMigrationError) Error() string {
|
||||||
|
if e.Table == "" {
|
||||||
|
return "SQLite 到 MySQL 迁移失败:" + e.Stage
|
||||||
|
}
|
||||||
|
return fmt.Sprintf("SQLite 到 MySQL 迁移失败:%s(表 %s)", e.Stage, e.Table)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *SQLiteMigrationError) Unwrap() error { return e.Cause }
|
||||||
|
|
||||||
|
type sqliteMigrationTableSpec struct {
|
||||||
|
name string
|
||||||
|
columns []string
|
||||||
|
primaryKey []string
|
||||||
|
}
|
||||||
|
|
||||||
|
// 顺序同时满足 MySQL 外键依赖;列清单故意不含 MySQL 生成列 current_client_id。
|
||||||
|
var sqliteMigrationTables = []sqliteMigrationTableSpec{
|
||||||
|
{name: "users", columns: []string{"user_id", "username", "password_hash", "role", "status", "last_login_at", "password_changed_at", "created_at", "updated_at"}, primaryKey: []string{"user_id"}},
|
||||||
|
{name: "clients", columns: []string{"client_id", "name", "device_address", "platform", "pdd_package", "capabilities", "last_seen_at", "created_at", "updated_at"}, primaryKey: []string{"client_id"}},
|
||||||
|
{name: "shopee_products", columns: []string{"goods_id", "title", "shopee_status", "main_sku_code", "pdd_goods_url", "pdd_goods_id", "created_at", "updated_at"}, primaryKey: []string{"goods_id"}},
|
||||||
|
{name: "pdd_products", columns: []string{"id", "goods_id", "url", "title", "shop_name", "skus_json", "collect_status", "collect_msg", "artifact_ref", "collected_at", "deleted_at", "created_at", "updated_at"}, primaryKey: []string{"id"}},
|
||||||
|
{name: "shopee_skus", columns: []string{"sku_id", "goods_id", "spec_raw", "color", "size", "advice", "parse_ok", "sku_code", "is_manual", "created_at", "updated_at"}, primaryKey: []string{"sku_id"}},
|
||||||
|
{name: "syb_orders", columns: []string{"syb_id", "order_no", "title", "shopee_goods_id", "shopee_sku_id", "product_spec", "quantity", "price_twd_cent", "image_url", "syb_data", "created_at", "updated_at"}, primaryKey: []string{"syb_id"}},
|
||||||
|
{name: "sku_mappings", columns: []string{"shopee_sku_id", "pdd_goods_id", "pdd_option_key", "pdd_options", "goods_id", "mapped_at", "mapped_by"}, primaryKey: []string{"shopee_sku_id", "pdd_goods_id"}},
|
||||||
|
{name: "tasks", columns: []string{"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"}, primaryKey: []string{"task_id"}},
|
||||||
|
{name: "task_claims", columns: []string{"task_id", "client_id", "claimed_at"}, primaryKey: []string{"task_id", "client_id"}},
|
||||||
|
{name: "idempotency_keys", columns: []string{"key", "request_hash", "response_body", "created_at"}, primaryKey: []string{"key"}},
|
||||||
|
{name: "syb_session", columns: []string{"username", "cookies", "expires_at", "updated_at"}, primaryKey: []string{"username"}},
|
||||||
|
{name: "syb_sync_state", columns: []string{"id", "last_synced_at", "updated_at"}, primaryKey: []string{"id"}},
|
||||||
|
{name: "web_sessions", columns: []string{"session_hash", "user_id", "expires_at", "created_at", "last_seen_at"}, primaryKey: []string{"session_hash"}},
|
||||||
|
{name: "client_user_assignments", columns: []string{"assignment_id", "client_id", "user_id", "started_at", "ended_at", "assigned_by_user_id", "ended_by_user_id", "end_reason"}, primaryKey: []string{"assignment_id"}},
|
||||||
|
{name: "syb_sync_runs", columns: []string{"run_id", "user_id", "date_from", "date_to", "status", "stock_count", "detail_count", "created_count", "updated_count", "skipped_count", "error_message", "cursor_advanced", "started_at", "finished_at"}, primaryKey: []string{"run_id"}},
|
||||||
|
}
|
||||||
|
|
||||||
|
// OpenLegacySQLiteReadOnly 以操作系统只读模式打开历史 admin.db。
|
||||||
|
func OpenLegacySQLiteReadOnly(path string) (*sql.DB, error) {
|
||||||
|
absPath, err := filepath.Abs(path)
|
||||||
|
if err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法解析 SQLite 路径", Cause: err}
|
||||||
|
}
|
||||||
|
dsn := "file:" + filepath.ToSlash(absPath) + "?mode=ro&_pragma=query_only(1)&_pragma=foreign_keys(1)"
|
||||||
|
db, err := sql.Open("sqlite", dsn)
|
||||||
|
if err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法准备只读 SQLite", Cause: err}
|
||||||
|
}
|
||||||
|
db.SetMaxOpenConns(1)
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := db.PingContext(ctx); err != nil {
|
||||||
|
db.Close()
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法打开只读 SQLite", Cause: err}
|
||||||
|
}
|
||||||
|
var queryOnly int
|
||||||
|
if err := db.QueryRow(`PRAGMA query_only`).Scan(&queryOnly); err != nil || queryOnly != 1 {
|
||||||
|
db.Close()
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "SQLite 未进入只读模式", Cause: err}
|
||||||
|
}
|
||||||
|
return db, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// MigrateSQLiteToMySQL 只允许把 SQLite v8 导入空的 MySQL 目标库。
|
||||||
|
func MigrateSQLiteToMySQL(source, target *sql.DB, dryRun bool) (*SQLiteMigrationSummary, error) {
|
||||||
|
if err := validateLegacySQLite(source); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := validateEmptyMySQLTarget(target); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
summary, err := migrationSummary(source, dryRun)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if dryRun {
|
||||||
|
return summary, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
tx, err := target.BeginTx(context.Background(), &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||||||
|
if err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法开始 MySQL 导入事务", Cause: err}
|
||||||
|
}
|
||||||
|
defer tx.Rollback()
|
||||||
|
for _, spec := range sqliteMigrationTables {
|
||||||
|
if err := copySQLiteTable(source, tx, spec); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := tx.Commit(); err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法提交 MySQL 导入事务", Cause: err}
|
||||||
|
}
|
||||||
|
if _, err := VerifySQLiteToMySQL(source, target); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return summary, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// VerifySQLiteToMySQL 比较逐表数量和全部迁移列的确定性摘要,不输出业务内容。
|
||||||
|
func VerifySQLiteToMySQL(source, target *sql.DB) (*SQLiteMigrationSummary, error) {
|
||||||
|
if err := validateLegacySQLite(source); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := CheckMySQLSchema(target); err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "MySQL schema 不完整", Cause: err}
|
||||||
|
}
|
||||||
|
summary, err := migrationSummary(source, false)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
for _, spec := range sqliteMigrationTables {
|
||||||
|
sourceCount, sourceDigest, err := tableDigest(source, spec, false)
|
||||||
|
if err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法核对 SQLite 数据", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
targetCount, targetDigest, err := tableDigest(target, spec, true)
|
||||||
|
if err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法核对 MySQL 数据", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
if sourceCount != targetCount || sourceDigest != targetDigest {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "数量或内容摘要不一致", Table: spec.name}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := verifyMySQLRelations(target); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return summary, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateLegacySQLite(db *sql.DB) error {
|
||||||
|
var version int
|
||||||
|
if err := db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != schemaVersion {
|
||||||
|
return &SQLiteMigrationError{Stage: "SQLite schema 必须是 v8", Cause: err}
|
||||||
|
}
|
||||||
|
var quickCheck string
|
||||||
|
if err := db.QueryRow(`PRAGMA quick_check`).Scan(&quickCheck); err != nil || quickCheck != "ok" {
|
||||||
|
return &SQLiteMigrationError{Stage: "SQLite 一致性检查失败", Cause: err}
|
||||||
|
}
|
||||||
|
rows, err := db.Query(`PRAGMA foreign_key_check`)
|
||||||
|
if err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "SQLite 外键检查失败", Cause: err}
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
if rows.Next() {
|
||||||
|
return &SQLiteMigrationError{Stage: "SQLite 存在外键异常"}
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "SQLite 外键检查失败", Cause: err}
|
||||||
|
}
|
||||||
|
if err := CheckSchema(db); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "SQLite schema 不完整", Cause: err}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateEmptyMySQLTarget(db *sql.DB) error {
|
||||||
|
if err := CheckMySQLSchema(db); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "MySQL schema 不完整", Cause: err}
|
||||||
|
}
|
||||||
|
for _, spec := range sqliteMigrationTables {
|
||||||
|
var count int64
|
||||||
|
if err := db.QueryRow("SELECT COUNT(*) FROM " + quoteMySQLIdentifier(spec.name)).Scan(&count); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法检查 MySQL 目标", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
if count != 0 {
|
||||||
|
return &SQLiteMigrationError{Stage: "MySQL 目标不是空库", Table: spec.name}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func migrationSummary(db *sql.DB, dryRun bool) (*SQLiteMigrationSummary, error) {
|
||||||
|
summary := &SQLiteMigrationSummary{DryRun: dryRun, Tables: make([]SQLiteMigrationTable, 0, len(sqliteMigrationTables))}
|
||||||
|
for _, spec := range sqliteMigrationTables {
|
||||||
|
var count int64
|
||||||
|
if err := db.QueryRow("SELECT COUNT(*) FROM " + quoteSQLiteIdentifier(spec.name)).Scan(&count); err != nil {
|
||||||
|
return nil, &SQLiteMigrationError{Stage: "无法统计 SQLite 数据", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
summary.Tables = append(summary.Tables, SQLiteMigrationTable{Name: spec.name, Count: count})
|
||||||
|
}
|
||||||
|
return summary, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func copySQLiteTable(source *sql.DB, target *sql.Tx, spec sqliteMigrationTableSpec) error {
|
||||||
|
selectSQL := "SELECT " + joinQuoted(spec.columns, quoteSQLiteIdentifier) + " FROM " + quoteSQLiteIdentifier(spec.name) +
|
||||||
|
" ORDER BY " + joinQuoted(spec.primaryKey, quoteSQLiteIdentifier)
|
||||||
|
rows, err := source.Query(selectSQL)
|
||||||
|
if err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法读取 SQLite", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(spec.columns)), ",")
|
||||||
|
insertSQL := "INSERT INTO " + quoteMySQLIdentifier(spec.name) + " (" +
|
||||||
|
joinQuoted(spec.columns, quoteMySQLIdentifier) + ") VALUES (" + placeholders + ")"
|
||||||
|
stmt, err := target.Prepare(insertSQL)
|
||||||
|
if err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法准备 MySQL 导入", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
defer stmt.Close()
|
||||||
|
for rows.Next() {
|
||||||
|
values, pointers := scanBuffers(len(spec.columns))
|
||||||
|
if err := rows.Scan(pointers...); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法读取 SQLite 行", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
if _, err := stmt.Exec(values...); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法写入 MySQL", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法遍历 SQLite", Table: spec.name, Cause: err}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func tableDigest(db *sql.DB, spec sqliteMigrationTableSpec, mysql bool) (int64, [32]byte, error) {
|
||||||
|
quote := quoteSQLiteIdentifier
|
||||||
|
if mysql {
|
||||||
|
quote = quoteMySQLIdentifier
|
||||||
|
}
|
||||||
|
query := "SELECT " + joinQuoted(spec.columns, quote) + " FROM " + quote(spec.name) +
|
||||||
|
" ORDER BY " + joinQuoted(spec.primaryKey, quote)
|
||||||
|
rows, err := db.Query(query)
|
||||||
|
if err != nil {
|
||||||
|
return 0, [32]byte{}, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
hash := sha256.New()
|
||||||
|
var count int64
|
||||||
|
for rows.Next() {
|
||||||
|
values, pointers := scanBuffers(len(spec.columns))
|
||||||
|
if err := rows.Scan(pointers...); err != nil {
|
||||||
|
return 0, [32]byte{}, err
|
||||||
|
}
|
||||||
|
for _, value := range values {
|
||||||
|
writeDigestValue(hash, value)
|
||||||
|
}
|
||||||
|
count++
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return 0, [32]byte{}, err
|
||||||
|
}
|
||||||
|
var digest [32]byte
|
||||||
|
copy(digest[:], hash.Sum(nil))
|
||||||
|
return count, digest, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func verifyMySQLRelations(db *sql.DB) error {
|
||||||
|
checks := []struct {
|
||||||
|
name string
|
||||||
|
query string
|
||||||
|
}{
|
||||||
|
{"SKU 商品关系", `SELECT COUNT(*) FROM shopee_skus s LEFT JOIN shopee_products p ON p.goods_id=s.goods_id WHERE p.goods_id IS NULL`},
|
||||||
|
{"规格映射关系", `SELECT COUNT(*) FROM sku_mappings m LEFT JOIN shopee_skus s ON s.sku_id=m.shopee_sku_id WHERE s.sku_id IS NULL`},
|
||||||
|
{"网页登录会话关系", `SELECT COUNT(*) FROM web_sessions s LEFT JOIN users u ON u.user_id=s.user_id WHERE u.user_id IS NULL`},
|
||||||
|
{"客户端归属用户关系", `SELECT COUNT(*) FROM client_user_assignments a LEFT JOIN users u ON u.user_id=a.user_id WHERE u.user_id IS NULL`},
|
||||||
|
{"同步记录用户关系", `SELECT COUNT(*) FROM syb_sync_runs r LEFT JOIN users u ON u.user_id=r.user_id WHERE u.user_id IS NULL`},
|
||||||
|
{"当前客户端归属唯一", `SELECT COUNT(*) FROM (SELECT client_id FROM client_user_assignments WHERE ended_at IS NULL GROUP BY client_id HAVING COUNT(*) > 1) duplicate_assignments`},
|
||||||
|
}
|
||||||
|
for _, check := range checks {
|
||||||
|
var count int64
|
||||||
|
if err := db.QueryRow(check.query).Scan(&count); err != nil {
|
||||||
|
return &SQLiteMigrationError{Stage: "无法核对" + check.name, Cause: err}
|
||||||
|
}
|
||||||
|
if count != 0 {
|
||||||
|
return &SQLiteMigrationError{Stage: check.name + "不一致"}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func scanBuffers(count int) ([]any, []any) {
|
||||||
|
values := make([]any, count)
|
||||||
|
pointers := make([]any, count)
|
||||||
|
for i := range values {
|
||||||
|
pointers[i] = &values[i]
|
||||||
|
}
|
||||||
|
return values, pointers
|
||||||
|
}
|
||||||
|
|
||||||
|
func writeDigestValue(hash interface{ Write([]byte) (int, error) }, value any) {
|
||||||
|
if value == nil {
|
||||||
|
hash.Write([]byte{0})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var bytes []byte
|
||||||
|
switch typed := value.(type) {
|
||||||
|
case []byte:
|
||||||
|
bytes = typed
|
||||||
|
case string:
|
||||||
|
bytes = []byte(typed)
|
||||||
|
case time.Time:
|
||||||
|
bytes = []byte(typed.UTC().Format(time.RFC3339Nano))
|
||||||
|
default:
|
||||||
|
bytes = []byte(fmt.Sprint(typed))
|
||||||
|
}
|
||||||
|
hash.Write([]byte{1})
|
||||||
|
var length [8]byte
|
||||||
|
binary.BigEndian.PutUint64(length[:], uint64(len(bytes)))
|
||||||
|
hash.Write(length[:])
|
||||||
|
hash.Write(bytes)
|
||||||
|
}
|
||||||
|
|
||||||
|
func joinQuoted(names []string, quote func(string) string) string {
|
||||||
|
quoted := make([]string, len(names))
|
||||||
|
for i, name := range names {
|
||||||
|
quoted[i] = quote(name)
|
||||||
|
}
|
||||||
|
return strings.Join(quoted, ",")
|
||||||
|
}
|
||||||
|
|
||||||
|
func quoteSQLiteIdentifier(name string) string {
|
||||||
|
return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func quoteMySQLIdentifier(name string) string {
|
||||||
|
return "`" + strings.ReplaceAll(name, "`", "``") + "`"
|
||||||
|
}
|
||||||
|
|
||||||
|
// SortedMigrationSummary 供命令稳定输出,避免依赖内部外键导入顺序。
|
||||||
|
func SortedMigrationSummary(summary *SQLiteMigrationSummary) []SQLiteMigrationTable {
|
||||||
|
tables := append([]SQLiteMigrationTable(nil), summary.Tables...)
|
||||||
|
sort.Slice(tables, func(i, j int) bool { return tables[i].Name < tables[j].Name })
|
||||||
|
return tables
|
||||||
|
}
|
||||||
@@ -0,0 +1,170 @@
|
|||||||
|
package repository
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/sha256"
|
||||||
|
"database/sql"
|
||||||
|
"errors"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"cmautobuy/admin/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestMigrateSQLiteToMySQL_演练迁移核对和重复保护(t *testing.T) {
|
||||||
|
sourcePath := newLegacyMigrationSource(t)
|
||||||
|
before := fileDigest(t, sourcePath)
|
||||||
|
source, err := OpenLegacySQLiteReadOnly(sourcePath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer source.Close()
|
||||||
|
target := newSybTestDB(t)
|
||||||
|
|
||||||
|
dryRun, err := MigrateSQLiteToMySQL(source, target, true)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("演练失败: %v", err)
|
||||||
|
}
|
||||||
|
if !dryRun.DryRun || summaryCount(dryRun, "users") != 1 || summaryCount(dryRun, "tasks") != 1 {
|
||||||
|
t.Fatalf("演练摘要不对: %+v", dryRun)
|
||||||
|
}
|
||||||
|
assertBusinessTablesEmpty(t, target)
|
||||||
|
|
||||||
|
result, err := MigrateSQLiteToMySQL(source, target, false)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("迁移失败: %v", err)
|
||||||
|
}
|
||||||
|
if result.DryRun {
|
||||||
|
t.Fatal("正式迁移摘要不应标记为 dry-run")
|
||||||
|
}
|
||||||
|
if _, err := VerifySQLiteToMySQL(source, target); err != nil {
|
||||||
|
t.Fatalf("迁移后核对失败: %v", err)
|
||||||
|
}
|
||||||
|
if got := fileDigest(t, sourcePath); got != before {
|
||||||
|
t.Fatal("迁移修改了源 SQLite 文件")
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := MigrateSQLiteToMySQL(source, target, false); err == nil {
|
||||||
|
t.Fatal("重复迁移到非空目标应该被拒绝")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMigrateSQLiteToMySQL_中途失败整体回滚(t *testing.T) {
|
||||||
|
sourcePath := newLegacyMigrationSource(t)
|
||||||
|
source, err := OpenLegacySQLiteReadOnly(sourcePath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer source.Close()
|
||||||
|
target := newSybTestDB(t)
|
||||||
|
if _, err := target.Exec(`CREATE TRIGGER reject_tasks BEFORE INSERT ON tasks
|
||||||
|
FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'fixture rejection'`); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = MigrateSQLiteToMySQL(source, target, false)
|
||||||
|
var migrationErr *SQLiteMigrationError
|
||||||
|
if !errors.As(err, &migrationErr) || migrationErr.Table != "tasks" {
|
||||||
|
t.Fatalf("应该返回不含业务数据的 tasks 迁移错误,实际 %v", err)
|
||||||
|
}
|
||||||
|
assertBusinessTablesEmpty(t, target)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMigrateSQLiteToMySQL_拒绝错误SQLite版本(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
db, err := Open(dir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := Migrate(db); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := db.Exec(`PRAGMA user_version = 7`); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
db.Close()
|
||||||
|
source, err := OpenLegacySQLiteReadOnly(filepath.Join(dir, "admin.db"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer source.Close()
|
||||||
|
target := newSybTestDB(t)
|
||||||
|
if _, err := MigrateSQLiteToMySQL(source, target, true); err == nil {
|
||||||
|
t.Fatal("错误 SQLite 版本应该被拒绝")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func newLegacyMigrationSource(t *testing.T) string {
|
||||||
|
t.Helper()
|
||||||
|
dir := t.TempDir()
|
||||||
|
db, err := Open(dir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := Migrate(db); err != nil {
|
||||||
|
db.Close()
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
now := model.NowISO()
|
||||||
|
statements := []struct {
|
||||||
|
query string
|
||||||
|
args []any
|
||||||
|
}{
|
||||||
|
{`INSERT INTO users (user_id,username,password_hash,role,status,password_changed_at,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?)`, []any{"USR-1", "admin", "bcrypt-fixture", "admin", "active", now, now, now}},
|
||||||
|
{`INSERT INTO clients (client_id,name,last_seen_at,created_at,updated_at) VALUES (?,?,?,?,?)`, []any{"CLIENT-1", "测试客户端", now, now, now}},
|
||||||
|
{`INSERT INTO shopee_products (goods_id,title,pdd_goods_url,pdd_goods_id,created_at,updated_at) VALUES (?,?,?,?,?,?)`, []any{"SP-1", "测试商品", "https://example.invalid/pdd", "PDD-1", now, now}},
|
||||||
|
{`INSERT INTO pdd_products (id,goods_id,url,title,skus_json,collect_status,created_at,updated_at) VALUES (?,?,?,?,?,'collected',?,?)`, []any{7, "PDD-1", "https://example.invalid/pdd", "PDD 商品", `{"skus":[]}`, now, now}},
|
||||||
|
{`INSERT INTO shopee_skus (sku_id,goods_id,spec_raw,color,size,parse_ok,created_at,updated_at) VALUES (?,?,?,?,?,1,?,?)`, []any{"SKU-1", "SP-1", "黑色,M", "黑色", "M", now, now}},
|
||||||
|
{`INSERT INTO syb_orders (syb_id,order_no,title,shopee_goods_id,shopee_sku_id,product_spec,quantity,syb_data,created_at,updated_at) VALUES (?,?,?,?,?,?,1,'{}',?,?)`, []any{"SYB-1", "ORDER-1", "测试货运单", "SP-1", "SKU-1", "黑色,M", now, now}},
|
||||||
|
{`INSERT INTO sku_mappings (shopee_sku_id,pdd_goods_id,pdd_option_key,pdd_options,goods_id,mapped_at,mapped_by) VALUES (?,?,?,?,?,?,?)`, []any{"SKU-1", "PDD-1", "color=黑色&size=M", `{"color":"黑色","size":"M"}`, "SP-1", now, "USR-1"}},
|
||||||
|
{`INSERT INTO tasks (task_id,task_type,status,assigned_client,syb_id,pdd_goods_url,pdd_goods_id,created_at,updated_at) VALUES (?,'collect','claimed',?,?,?,?,?,?)`, []any{"TASK-1", "CLIENT-1", "SYB-1", "https://example.invalid/pdd", "PDD-1", now, now}},
|
||||||
|
{`INSERT INTO task_claims (task_id,client_id,claimed_at) VALUES (?,?,?)`, []any{"TASK-1", "CLIENT-1", now}},
|
||||||
|
{`INSERT INTO idempotency_keys (key,request_hash,response_body,created_at) VALUES (?,?,?,?)`, []any{"IDEM-1", "hash", `{"accepted":true}`, now}},
|
||||||
|
{`INSERT INTO syb_session (username,cookies,expires_at,updated_at) VALUES (?,?,?,?)`, []any{"fixture", `[{"name":"session"}]`, now, now}},
|
||||||
|
{`INSERT INTO syb_sync_state (id,last_synced_at,updated_at) VALUES (1,?,?)`, []any{now, now}},
|
||||||
|
{`INSERT INTO web_sessions (session_hash,user_id,expires_at,created_at,last_seen_at) VALUES (?,?,?,?,?)`, []any{"SESSION-HASH", "USR-1", now, now, now}},
|
||||||
|
{`INSERT INTO client_user_assignments (assignment_id,client_id,user_id,started_at,assigned_by_user_id) VALUES (?,?,?,?,?)`, []any{"ASSIGN-1", "CLIENT-1", "USR-1", now, "USR-1"}},
|
||||||
|
{`INSERT INTO syb_sync_runs (run_id,user_id,date_from,date_to,status,started_at,finished_at) VALUES (?,?,?,?,'succeeded',?,?)`, []any{"RUN-1", "USR-1", "2026-08-09", "2026-08-09", now, now}},
|
||||||
|
}
|
||||||
|
for _, statement := range statements {
|
||||||
|
if _, err := db.Exec(statement.query, statement.args...); err != nil {
|
||||||
|
db.Close()
|
||||||
|
t.Fatalf("准备迁移 fixture 失败: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := db.Close(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return filepath.Join(dir, "admin.db")
|
||||||
|
}
|
||||||
|
|
||||||
|
func assertBusinessTablesEmpty(t *testing.T, db *sql.DB) {
|
||||||
|
t.Helper()
|
||||||
|
for _, spec := range sqliteMigrationTables {
|
||||||
|
var count int
|
||||||
|
if err := db.QueryRow("SELECT COUNT(*) FROM " + quoteMySQLIdentifier(spec.name)).Scan(&count); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if count != 0 {
|
||||||
|
t.Fatalf("表 %s 应为空,实际 %d", spec.name, count)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func summaryCount(summary *SQLiteMigrationSummary, name string) int64 {
|
||||||
|
for _, table := range summary.Tables {
|
||||||
|
if table.Name == name {
|
||||||
|
return table.Count
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return -1
|
||||||
|
}
|
||||||
|
|
||||||
|
func fileDigest(t *testing.T, path string) [32]byte {
|
||||||
|
t.Helper()
|
||||||
|
content, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return sha256.Sum256(content)
|
||||||
|
}
|
||||||
@@ -50,6 +50,7 @@
|
|||||||
| 加字段、改表、写 SQL | [03 数据模型](admin/03-data-model.md) | [02 架构](admin/02-architecture.md) §2 分层 |
|
| 加字段、改表、写 SQL | [03 数据模型](admin/03-data-model.md) | [02 架构](admin/02-architecture.md) §2 分层 |
|
||||||
| 改 Excel 导入 | [03 数据模型](admin/03-data-model.md) §3.3 | [00 术语表](admin/00-glossary.md) §3 upsert |
|
| 改 Excel 导入 | [03 数据模型](admin/03-data-model.md) §3.3 | [00 术语表](admin/00-glossary.md) §3 upsert |
|
||||||
| 改顺运宝同步 | [08 顺运宝接口](admin/08-顺运宝接口.md) | [03 数据模型](admin/03-data-model.md) |
|
| 改顺运宝同步 | [08 顺运宝接口](admin/08-顺运宝接口.md) | [03 数据模型](admin/03-data-model.md) |
|
||||||
|
| 把旧 Admin 数据迁移到 MySQL | [09 SQLite 单向迁移](admin/09-sqlite迁移到mysql.md) | [06 质量与安全](admin/06-quality-security.md) |
|
||||||
| 改给 Client 的接口 | [04 Client 接口实现](admin/04-client-api.md) | [Client 侧契约](client/04-admin-api-contract.md) |
|
| 改给 Client 的接口 | [04 Client 接口实现](admin/04-client-api.md) | [Client 侧契约](client/04-admin-api-contract.md) |
|
||||||
| 和 Admin 联调、登记新设备 | [07 设备登记联调手册](admin/07-设备登记联调手册.md) | [Client 侧契约](client/04-admin-api-contract.md) §5 |
|
| 和 Admin 联调、登记新设备 | [07 设备登记联调手册](admin/07-设备登记联调手册.md) | [Client 侧契约](client/04-admin-api-contract.md) §5 |
|
||||||
| 写测试 | [06 质量与安全](admin/06-quality-security.md) §2 | — |
|
| 写测试 | [06 质量与安全](admin/06-quality-security.md) §2 | — |
|
||||||
@@ -90,6 +91,7 @@
|
|||||||
| [06 质量、安全与测试](admin/06-quality-security.md) | 测试、Web 安全、发布门禁 |
|
| [06 质量、安全与测试](admin/06-quality-security.md) | 测试、Web 安全、发布门禁 |
|
||||||
| [07 设备登记联调手册](admin/07-设备登记联调手册.md) | **给 Client 开发者**:怎么让新设备登记成功 |
|
| [07 设备登记联调手册](admin/07-设备登记联调手册.md) | **给 Client 开发者**:怎么让新设备登记成功 |
|
||||||
| [08 顺运宝接口](admin/08-顺运宝接口.md) | 从抓包还原的外部 ERP 契约:登录、会话、货运单列表与明细 |
|
| [08 顺运宝接口](admin/08-顺运宝接口.md) | 从抓包还原的外部 ERP 契约:登录、会话、货运单列表与明细 |
|
||||||
|
| [09 SQLite 单向迁移](admin/09-sqlite迁移到mysql.md) | 只读演练、一次性导入、逐表核对和失败处理 |
|
||||||
|
|
||||||
## 文档标注说明
|
## 文档标注说明
|
||||||
|
|
||||||
|
|||||||
@@ -184,7 +184,8 @@ admin/data/
|
|||||||
跑源码时在 `admin\data\`;将来打包成 exe 后,在 exe 旁边。
|
跑源码时在 `admin\data\`;将来打包成 exe 后,在 exe 旁边。
|
||||||
|
|
||||||
每张表什么意思见 [03 数据模型](03-data-model.md)。旧 `admin.db` 只能通过
|
每张表什么意思见 [03 数据模型](03-data-model.md)。旧 `admin.db` 只能通过
|
||||||
单向迁移命令读取,生产 Admin 不再直接打开它。
|
单向迁移命令读取,生产 Admin 不再直接打开它。具体操作见
|
||||||
|
[09 SQLite 单向迁移到 MySQL 8](09-sqlite迁移到mysql.md)。
|
||||||
|
|
||||||
> 写代码时**不要自己拼这个路径**,用统一的 `DataDir()` 函数,
|
> 写代码时**不要自己拼这个路径**,用统一的 `DataDir()` 函数,
|
||||||
> 见 [02 架构](02-architecture.md) §6。
|
> 见 [02 架构](02-architecture.md) §6。
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
# SQLite 单向迁移到 MySQL 8
|
||||||
|
|
||||||
|
这份文档用于把历史 `admin.db` 一次性迁移到新的空 MySQL 8 数据库。迁移不是同步:成功切换后只使用 MySQL,不再把数据写回 SQLite。
|
||||||
|
|
||||||
|
## 1. 安全边界
|
||||||
|
|
||||||
|
- 源 `admin.db` 必须是 schema v8;工具使用操作系统只读模式和 SQLite `query_only` 打开。
|
||||||
|
- 目标必须是本程序刚建立且业务表为空的独立 MySQL 数据库;任何业务表非空都会拒绝执行。
|
||||||
|
- 不支持覆盖、合并、双写和重复导入。重新演练时新建另一个空测试库。
|
||||||
|
- 数据库密码只放环境变量,不要写入命令历史、仓库、日志或工单。
|
||||||
|
- 输出只包含表名和数量,不包含账号、Cookie、Session、订单号或商品明细。
|
||||||
|
|
||||||
|
## 2. 迁移前准备
|
||||||
|
|
||||||
|
先停止旧 Admin 的写入,复制一份只读备份,并记录哈希。以下命令从 `admin/` 目录执行:
|
||||||
|
|
||||||
|
```powershell
|
||||||
|
New-Item -ItemType Directory -Force data\backup
|
||||||
|
Copy-Item data\admin.db data\backup\admin-before-mysql.db
|
||||||
|
Get-FileHash data\admin.db -Algorithm SHA256
|
||||||
|
```
|
||||||
|
|
||||||
|
由数据库管理员提前创建独立的空数据库和最小权限账号。按部署环境设置这五个变量:
|
||||||
|
|
||||||
|
```powershell
|
||||||
|
$env:CMAUTOBUY_DB_HOST="127.0.0.1"
|
||||||
|
$env:CMAUTOBUY_DB_PORT="3307"
|
||||||
|
$env:CMAUTOBUY_DB_NAME="<空数据库名>"
|
||||||
|
$env:CMAUTOBUY_DB_USER="<迁移账号>"
|
||||||
|
$env:CMAUTOBUY_DB_PASSWORD="<从安全渠道取得>"
|
||||||
|
```
|
||||||
|
|
||||||
|
## 3. 先演练
|
||||||
|
|
||||||
|
```powershell
|
||||||
|
go run ./cmd/migrate-sqlite-to-mysql --source data/admin.db --dry-run
|
||||||
|
```
|
||||||
|
|
||||||
|
预期第一行是“演练检查通过,未写入数据”,后面只显示 15 张业务表的数量。此时 MySQL 业务表仍为空。
|
||||||
|
|
||||||
|
## 4. 执行并独立复核
|
||||||
|
|
||||||
|
```powershell
|
||||||
|
go run ./cmd/migrate-sqlite-to-mysql --source data/admin.db --execute
|
||||||
|
go run ./cmd/migrate-sqlite-to-mysql --source data/admin.db --verify-only
|
||||||
|
Get-FileHash data\admin.db -Algorithm SHA256
|
||||||
|
```
|
||||||
|
|
||||||
|
`--execute` 在一个 MySQL 事务中导入全部业务表,然后比较源、目标的逐表数量和全列摘要。`--verify-only` 再独立读取两边核对一次。最后的 SQLite 哈希应与迁移前一致。
|
||||||
|
|
||||||
|
## 5. 失败时怎样处理
|
||||||
|
|
||||||
|
- “SQLite schema 必须是 v8”:先用旧版本 Admin 的迁移逻辑把备份副本升级到 v8,不要直接改 `user_version`。
|
||||||
|
- “MySQL 目标不是空库”:停止操作,换一个新的空数据库;不要手工清表后继续生产迁移。
|
||||||
|
- “无法写入 MySQL”或“数量或内容摘要不一致”:不要启动新 Admin;保留源库和迁移前备份,丢弃本次目标库,修复原因后在新的空库重演。
|
||||||
|
- 迁移失败会回滚业务数据;MySQL schema 建表是可重放的,但业务数据绝不能部分保留后继续补写。
|
||||||
|
|
||||||
|
## 6. 切换门禁
|
||||||
|
|
||||||
|
只有下面各项都完成后才能让生产 Admin 使用新库:
|
||||||
|
|
||||||
|
- 演练、正式迁移和 `--verify-only` 均通过;
|
||||||
|
- SQLite 文件哈希迁移前后相同;
|
||||||
|
- MySQL 已备份,Admin 环境变量由系统服务安全注入;
|
||||||
|
- 页面登录、蝦皮/PDD/顺运宝列表、客户端领取和一次演练模式采购任务冒烟通过;
|
||||||
|
- 旧 `admin.db` 保留为只读回退证据,不再由 Admin 打开。
|
||||||
Reference in New Issue
Block a user