81 lines
2.1 KiB
Go
81 lines
2.1 KiB
Go
|
package store
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"fmt"
|
||
|
|
"io/fs"
|
||
|
|
"sort"
|
||
|
|
"strings"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
|
||
|
|
"github.com/hhs/camtalk/internal/logger"
|
||
|
|
)
|
||
|
|
|
||
|
|
// RunMigrations 从给定的 fs.FS 中读取 *.up.sql 文件并按版本号顺序执行。
|
||
|
|
// 已执行过的版本会跳过（通过 schema_migrations 表记录）。
|
||
|
|
func RunMigrations(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS) error {
|
||
|
|
// 确保 schema_migrations 表存在
|
||
|
|
if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
|
||
|
|
version INTEGER PRIMARY KEY,
|
||
|
|
applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||
|
|
)`); err != nil {
|
||
|
|
return fmt.Errorf("create schema_migrations table: %w", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// 收集所有 *.up.sql 文件
|
||
|
|
entries, err := fs.ReadDir(fsys, ".")
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("read migrations dir: %w", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
var files []string
|
||
|
|
for _, e := range entries {
|
||
|
|
if !e.IsDir() && strings.HasSuffix(e.Name(), ".up.sql") {
|
||
|
|
files = append(files, e.Name())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
sort.Strings(files)
|
||
|
|
|
||
|
|
for _, name := range files {
|
||
|
|
// 从文件名提取版本号，如 "001_users.up.sql" → 1
|
||
|
|
var version int
|
||
|
|
if _, err := fmt.Sscanf(name, "%d_", &version); err != nil {
|
||
|
|
return fmt.Errorf("parse version from %s: %w", name, err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// 检查是否已执行
|
||
|
|
var exists bool
|
||
|
|
if err := pool.QueryRow(ctx,
|
||
|
|
`SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, version,
|
||
|
|
).Scan(&exists); err != nil {
|
||
|
|
return fmt.Errorf("check migration version %d: %w", version, err)
|
||
|
|
}
|
||
|
|
if exists {
|
||
|
|
logger.Log.Debugw("migration already applied", "version", version, "file", name)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
|
||
|
|
// 读取并执行
|
||
|
|
content, err := fs.ReadFile(fsys, name)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("read migration %s: %w", name, err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if _, err := pool.Exec(ctx, string(content)); err != nil {
|
||
|
|
return fmt.Errorf("execute migration %s: %w", name, err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// 记录已执行
|
||
|
|
if _, err := pool.Exec(ctx,
|
||
|
|
`INSERT INTO schema_migrations (version) VALUES ($1)`, version,
|
||
|
|
); err != nil {
|
||
|
|
return fmt.Errorf("record migration %d: %w", version, err)
|
||
|
|
}
|
||
|
|
|
||
|
|
logger.Log.Infow("migration applied", "version", version, "file", name)
|
||
|
|
}
|
||
|
|
|
||
|
|
return nil
|
||
|
|
}
|