407 lines
16 KiB
Go
407 lines
16 KiB
Go
package repository
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/binary"
|
||
"encoding/json"
|
||
"fmt"
|
||
"path/filepath"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"cmautobuy/admin/spec"
|
||
)
|
||
|
||
// 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", "shop_name", "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 := normalizeMigratedShopeeSKUs(tx); 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
|
||
}
|
||
|
||
// normalizeMigratedShopeeSKUs 补齐 MySQL v8 新增、历史 SQLite 没有的规格身份列。
|
||
func normalizeMigratedShopeeSKUs(tx *sql.Tx) error {
|
||
rows, err := tx.Query(`SELECT sku_id,spec_raw,color,size,advice,sku_code,updated_at FROM shopee_skus`)
|
||
if err != nil {
|
||
return &SQLiteMigrationError{Stage: "无法读取待转换 SKU", Table: "shopee_skus", Cause: err}
|
||
}
|
||
type item struct {
|
||
id, raw, updated string
|
||
color, size, advice, code sql.NullString
|
||
}
|
||
var list []item
|
||
for rows.Next() {
|
||
var row item
|
||
if err := rows.Scan(&row.id, &row.raw, &row.color, &row.size, &row.advice, &row.code, &row.updated); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
list = append(list, row)
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
return err
|
||
}
|
||
for _, row := range list {
|
||
key, err := spec.SpecKey(row.raw)
|
||
if err != nil {
|
||
return &SQLiteMigrationError{Stage: "历史 SKU 规格原文无效", Table: "shopee_skus", Cause: err}
|
||
}
|
||
if len([]rune(key)) > 191 {
|
||
return &SQLiteMigrationError{Stage: "历史 SKU 规格键过长", Table: "shopee_skus"}
|
||
}
|
||
sources, times := map[string]string{}, map[string]string{}
|
||
for name, value := range map[string]string{"color": row.color.String, "size": row.size.String, "advice": row.advice.String, "sku_code": row.code.String} {
|
||
if value != "" {
|
||
sources[name] = "report"
|
||
times[name] = row.updated
|
||
}
|
||
}
|
||
sourcesJSON, _ := json.Marshal(sources)
|
||
timesJSON, _ := json.Marshal(times)
|
||
if _, err := tx.Exec(`UPDATE shopee_skus SET shopee_sku_id=sku_id,spec_key=?,source='report',field_sources=?,field_observed_at=? WHERE sku_id=?`, key, string(sourcesJSON), string(timesJSON), row.id); err != nil {
|
||
return &SQLiteMigrationError{Stage: "无法补齐 MySQL v8 SKU 身份", Table: "shopee_skus", Cause: err}
|
||
}
|
||
}
|
||
return 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
|
||
}
|