Files
cmautobuy/admin/repository/db.go
T

215 lines
7.5 KiB
Go
Raw Normal View History

// Package repository 封装 SQLite 读写。
//
// 改动本文件前必读 admin/AGENTS.md。三条硬规则:
// - 只有本包能写 SQL,handler 和 service 都不许拼 SQL;
// - SQL 一律参数化查询(用 ? 占位),禁止字符串拼接;
// - 迁移只能往前加,不许在启动时删库重建——data/ 在升级时是保留的,
// 里面有人工填了几个月的 PDD 链接和 SKU 映射。
//
// 表结构的权威定义在 docs/admin/03-data-model.md,改表要先改文档。
package repository
import (
"database/sql"
"fmt"
"path/filepath"
// 纯 Go 的 SQLite 驱动,注册的驱动名是 "sqlite"(不是 "sqlite3")。
// 不得换成 github.com/mattn/go-sqlite3,那个需要 cgo,
// Windows 上要装 gcc,打包 exe 会变麻烦。理由见 admin/AGENTS.md。
_ "modernc.org/sqlite"
)
// Open 打开 data/admin.db,并设置必要的 PRAGMA。
func Open(dataDir string) (*sql.DB, error) {
path := filepath.Join(dataDir, "admin.db")
db, err := sql.Open("sqlite", path)
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)
}
}
return db, nil
}
// migrations 按顺序存放每一版的迁移语句。
//
// 外层一个元素 = 一个版本;内层是该版本要执行的语句,**一条一执行**。
// 不把多条语句塞进一个字符串,是因为 database/sql 的 Exec 对
// "一次执行多条语句"的支持因驱动而异,拆开最稳妥。
//
// 加新版本时**只能往末尾追加**,不许改动已有元素——
// 已经发布出去的库是按旧语句建的,改了会导致新旧库结构不一致。
var migrations = [][]string{
// v1: 初始表结构,对应 docs/admin/03-data-model.md
{
`CREATE TABLE shopee_products (
goods_id TEXT PRIMARY KEY,
title TEXT NOT NULL,
shopee_status TEXT,
main_sku_code TEXT,
-- 下面三个是人工维护的,报表里没有,导入时绝不能覆盖
pdd_goods_url TEXT,
pdd_goods_id TEXT,
pdd_data TEXT,
collect_status TEXT NOT NULL DEFAULT 'no_link'
CHECK (collect_status IN (
'no_link', 'pending', 'collecting',
'collected', 'failed'
)),
collect_error TEXT,
collected_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);`,
`CREATE INDEX idx_shopee_products_status ON shopee_products(collect_status);`,
`CREATE TABLE shopee_skus (
sku_id TEXT PRIMARY KEY,
goods_id TEXT NOT NULL,
spec_raw TEXT NOT NULL,
color TEXT,
size TEXT,
advice TEXT,
parse_ok INTEGER NOT NULL DEFAULT 0,
sku_code TEXT,
is_manual INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (goods_id) REFERENCES shopee_products(goods_id) ON DELETE CASCADE
);`,
`CREATE INDEX idx_shopee_skus_goods ON shopee_skus(goods_id);`,
`CREATE INDEX idx_shopee_skus_parse ON shopee_skus(parse_ok);`,
`CREATE TABLE syb_orders (
syb_id TEXT PRIMARY KEY,
order_no TEXT NOT NULL,
title TEXT,
shopee_goods_id TEXT,
shopee_sku_id TEXT,
quantity INTEGER NOT NULL CHECK (quantity > 0),
price_twd_cent INTEGER CHECK (price_twd_cent IS NULL OR price_twd_cent >= 0),
image_url TEXT,
syb_data TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);`,
`CREATE INDEX idx_syb_orders_order ON syb_orders(order_no);`,
`CREATE INDEX idx_syb_orders_goods ON syb_orders(shopee_goods_id);`,
`CREATE INDEX idx_syb_orders_list ON syb_orders(updated_at DESC, syb_id DESC);`,
`CREATE TABLE sku_mappings (
shopee_sku_id TEXT PRIMARY KEY,
goods_id TEXT NOT NULL,
pdd_options TEXT NOT NULL,
mapped_at TEXT NOT NULL,
mapped_by TEXT,
FOREIGN KEY (shopee_sku_id) REFERENCES shopee_skus(sku_id) ON DELETE CASCADE
);`,
`CREATE INDEX idx_sku_mappings_goods ON sku_mappings(goods_id);`,
`CREATE TABLE tasks (
task_id TEXT PRIMARY KEY,
task_type TEXT NOT NULL CHECK (task_type IN ('collect', 'purchase')),
status TEXT NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'assigned', 'claimed',
'succeeded', 'manual_review',
'failed', 'cancelled')),
version INTEGER NOT NULL DEFAULT 1 CHECK (version > 0),
priority INTEGER NOT NULL DEFAULT 0,
assigned_client TEXT,
claimed_at TEXT,
syb_id TEXT,
order_no TEXT,
goods_id TEXT,
shopee_sku_id TEXT,
-- Client 契约要求:pdd_goods_url 必填;
-- 采购任务的 quantity 和 max_price_cent 也必填(价格保护)
pdd_goods_url TEXT NOT NULL,
pdd_goods_id TEXT,
pdd_options TEXT,
quantity INTEGER CHECK (quantity IS NULL OR quantity > 0),
max_price_cent INTEGER CHECK (max_price_cent IS NULL OR max_price_cent > 0),
result_data TEXT,
error_code TEXT,
error_message TEXT,
finished_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);`,
`CREATE INDEX idx_tasks_claim ON tasks(assigned_client, status, priority DESC, created_at);`,
`CREATE INDEX idx_tasks_list ON tasks(updated_at DESC, task_id DESC);`,
`CREATE INDEX idx_tasks_order ON tasks(order_no);`,
`CREATE TABLE clients (
client_id TEXT PRIMARY KEY,
name TEXT,
device_address TEXT,
platform TEXT,
pdd_package TEXT,
capabilities TEXT,
last_seen_at TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);`,
`CREATE TABLE idempotency_keys (
key TEXT PRIMARY KEY,
request_hash TEXT NOT NULL,
response_body TEXT NOT NULL,
created_at TEXT NOT NULL
);`,
},
}
// Migrate 把数据库升到最新版本。
// 已经是最新的就什么都不做,可以重复调用。
func Migrate(db *sql.DB) error {
var current int
if err := db.QueryRow("PRAGMA user_version").Scan(&current); err != nil {
return fmt.Errorf("读取 user_version 失败: %w", err)
}
if current > len(migrations) {
return fmt.Errorf(
"数据库版本 %d 高于本程序支持的 %d,"+
"说明这个库是更新版本的程序建的,请升级程序而不是降级",
current, len(migrations))
}
for v := current; v < len(migrations); v++ {
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("开始迁移 v%d 失败: %w", v+1, err)
}
for i, stmt := range migrations[v] {
if _, err := tx.Exec(stmt); err != nil {
tx.Rollback()
return fmt.Errorf("执行迁移 v%d 第 %d 条语句失败: %w", v+1, i+1, err)
}
}
// PRAGMA 不支持参数化,这里的值来自循环变量而非外部输入,安全。
if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", v+1)); err != nil {
tx.Rollback()
return fmt.Errorf("更新 user_version 到 %d 失败: %w", v+1, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("提交迁移 v%d 失败: %w", v+1, err)
}
}
return nil
}