update

cmd/dbtool/main.go

@@ -62,6 +62,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"path/filepath"
 	"sort"
 	"strings"
 	"time"
@@ -111,6 +112,8 @@ func main() {
 		runFixDup(os.Args[2:])
 	case "purge-rejected":
 		runPurgeRejected(os.Args[2:])
+	case "migrate":
+		runMigrate(os.Args[2:])
 	default:
 		usage()
 		os.Exit(2)
@@ -131,6 +134,11 @@ func usage() {
       把现存「已驳回（rejected）」且未下架的记录转删除：级联软删其全部图片、记录置
       is_deleted=1（后台即消失），并把图片 key 入队 media_cleanup，由运行中的 worker
       按引用计数清理 S4 孤儿文件（与后台点「删除（清空 S4）」等价，但一次性全局处理）
+  dbtool migrate [-config <yml] [-in <file.sql>]
+      执行 db/migrations 下的迁移脚本（按文件名排序、逐文件执行、幂等可重跑）：
+      每个文件执行成功后记入 schema_migrations，重跑已记录的会跳过；库内「已存在」
+      视为已应用（也记入记录表后跳过），因此历史已手动 psql 过的迁移重跑不会报错中断。
+      -in 指定单个文件时，只执行该文件。本子命令替代 psql，免装客户端即可落库。
 
 说明:
   连接信息读取 configs/config.yml 的 database 段（可用 -config 覆盖）。
@@ -405,6 +413,110 @@ func runPurgeRejected(args []string) {
 	log.Printf("✓ 完成：转删除记录 %d 条、图片 %d 张，已入队 media_cleanup（运行中 worker 将按引用计数清理 S4）", totalRecs, totalKeys)
 }
 
+// runMigrate 执行 db/migrations 下的迁移脚本（按文件名排序、逐文件执行、幂等可重跑）。
+//
+// 每个文件执行成功后记入 schema_migrations(name)，重跑已记录的会直接跳过；
+// 库内「已存在」（如 ALTER ADD COLUMN 报 column already exists，说明该迁移早先被
+// 手动 psql 执行过）也视作已应用，记入记录表后跳过——这样历史已手动落库的迁移重跑
+// 不会报错中断，新迁移则正常执行。
+//
+// 用现有的 execScript（pgx 简单查询协议，整段脚本在一个隐式事务里执行）跑，免装 psql。
+// 默认执行 db/migrations/*.sql；-in 指定单文件时只执行该文件。
+func runMigrate(args []string) {
+	fs := flag.NewFlagSet("migrate", flag.ExitOnError)
+	in := fs.String("in", "", "只执行单个迁移文件（默认执行 db/migrations 下全部 *.sql，按文件名排序）")
+	cfgPath := fs.String("config", "", "配置文件路径（默认 configs/config.yml）")
+	_ = fs.Parse(args)
+
+	db, dcfg, err := openDB(*cfgPath)
+	if err != nil {
+		log.Fatalf("✗ %v", err)
+	}
+	defer db.Close()
+
+	if err := ensureMigrationsTable(db); err != nil {
+		log.Fatalf("✗ 建迁移记录表失败: %v", err)
+	}
+
+	files, err := migrationFiles(*in)
+	if err != nil {
+		log.Fatalf("✗ %v", err)
+	}
+	if len(files) == 0 {
+		log.Fatalf("✗ 没有待执行迁移（检查 db/migrations 目录或 -in 路径）")
+	}
+
+	for _, path := range files {
+		name := filepath.Base(path)
+		applied, err := isMigrationApplied(db, name)
+		if err != nil {
+			log.Fatalf("✗ 查询迁移记录失败: %v", err)
+		}
+		if applied {
+			log.Printf("→ 跳过（已应用）: %s", name)
+			continue
+		}
+		script, err := os.ReadFile(path)
+		if err != nil {
+			log.Fatalf("✗ 读迁移文件 %s 失败: %v", path, err)
+		}
+		log.Printf("→ 执行迁移: %s", name)
+		if err := execScript(context.Background(), db, string(script)); err != nil {
+			if strings.Contains(err.Error(), "already exists") {
+				log.Printf("→ 跳过（库内已存在，视为已应用）: %s", name)
+				if rerr := recordMigration(db, name); rerr != nil {
+					log.Fatalf("✗ 记录迁移失败: %v", rerr)
+				}
+				continue
+			}
+			log.Fatalf("✗ 迁移失败 %s（已整体回滚）: %v", name, err)
+		}
+		if err := recordMigration(db, name); err != nil {
+			log.Fatalf("✗ 记录迁移失败: %v", err)
+		}
+		log.Printf("✓ 已应用: %s", name)
+	}
+	log.Printf("✓ 迁移完成（%s）", dcfg.Addr())
+}
+
+// migrationFiles 返回待执行的迁移文件列表（按文件名排序）。-in 指定单文件时只返回它。
+func migrationFiles(in string) ([]string, error) {
+	if in != "" {
+		return []string{in}, nil
+	}
+	matches, err := filepath.Glob("db/migrations/*.sql")
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(matches)
+	return matches, nil
+}
+
+// ensureMigrationsTable 建迁移记录表（幂等）。
+func ensureMigrationsTable(db *sql.DB) error {
+	_, err := db.Exec(`CREATE TABLE IF NOT EXISTS public.schema_migrations (
+		name text PRIMARY KEY,
+		applied_at timestamptz NOT NULL DEFAULT now()
+	)`)
+	return err
+}
+
+// isMigrationApplied 该迁移是否已记录（已成功执行）。
+func isMigrationApplied(db *sql.DB, name string) (bool, error) {
+	var n int
+	err := db.QueryRow(`SELECT count(*) FROM public.schema_migrations WHERE name = $1`, name).Scan(&n)
+	if err != nil {
+		return false, err
+	}
+	return n > 0, nil
+}
+
+// recordMigration 把迁移记入记录表（幂等，重跑不报错）。
+func recordMigration(db *sql.DB, name string) error {
+	_, err := db.Exec(`INSERT INTO public.schema_migrations (name) VALUES ($1) ON CONFLICT (name) DO NOTHING`, name)
+	return err
+}
+
 // selectUint32s 执行返回单列 uint32 的查询。
 func selectUint32s(db *sql.DB, q string, args ...any) ([]uint32, error) {
 	rows, err := db.Query(q, args...)