merge: config + store(UTC DSN) + 自动迁移 [tsk_zRA6fGU1JuHj]
以 main 现有 config 设计为准(FromEnv/*Config/单 DSN/RS256),移植分支新增的 store 层:mysql.go 的 Open(*config.Config) 改用 cfg.DSN、结构化强制 UTC + 会话时区断言;migrate.go + migrations/embed.go 提供 golang-migrate 嵌入式迁移;cmd/migrate/main.go 实现 migrate CLI(读 DB_DSN)。未采用分支的 Load()/PANGOLIN_ 前缀/MySQL 拆分字段(与 main 设计冲突)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
@@ -1,14 +1,73 @@
|
||||
// Command migrate applies or reverts database schema migrations for the
|
||||
// Pangolin server.
|
||||
//
|
||||
// Usage:
|
||||
//
|
||||
// migrate up – apply all pending migrations
|
||||
// migrate down – revert all applied migrations
|
||||
// migrate version – print the current migration version
|
||||
//
|
||||
// Configuration is read from the DB_DSN environment variable (a full MySQL
|
||||
// DSN), consistent with the rest of the server.
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/config"
|
||||
"github.com/wangjia/pangolin/server/internal/store"
|
||||
)
|
||||
|
||||
func main() {
|
||||
fmt.Fprintln(os.Stdout, "pangolin-migrate: database migration tool")
|
||||
fmt.Fprintln(os.Stdout, "Usage: migrate [-up|-down] [-steps N]")
|
||||
fmt.Fprintln(os.Stdout, "")
|
||||
fmt.Fprintln(os.Stdout, " (implementation in task 1e)")
|
||||
os.Exit(0)
|
||||
if len(os.Args) < 2 {
|
||||
usage()
|
||||
os.Exit(1)
|
||||
}
|
||||
cmd := os.Args[1]
|
||||
|
||||
dsn := os.Getenv("DB_DSN")
|
||||
if dsn == "" {
|
||||
log.Fatal("migrate: DB_DSN is required")
|
||||
}
|
||||
|
||||
db, err := store.Open(&config.Config{DSN: dsn})
|
||||
if err != nil {
|
||||
log.Fatalf("migrate: db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
switch cmd {
|
||||
case "up":
|
||||
if err := store.MigrateUp(db); err != nil {
|
||||
log.Fatalf("migrate up: %v", err)
|
||||
}
|
||||
log.Println("migrate: up — done")
|
||||
|
||||
case "down":
|
||||
if err := store.MigrateDown(db); err != nil {
|
||||
log.Fatalf("migrate down: %v", err)
|
||||
}
|
||||
log.Println("migrate: down — done")
|
||||
|
||||
case "version":
|
||||
version, dirty, err := store.MigrateVersion(db)
|
||||
if err != nil {
|
||||
log.Fatalf("migrate version: %v", err)
|
||||
}
|
||||
fmt.Printf("migrate: version=%d dirty=%v\n", version, dirty)
|
||||
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "migrate: unknown command %q\n", cmd)
|
||||
usage()
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func usage() {
|
||||
fmt.Fprintln(os.Stderr, "Usage: migrate <up|down|version>")
|
||||
fmt.Fprintln(os.Stderr, "")
|
||||
fmt.Fprintln(os.Stderr, "Environment variables:")
|
||||
fmt.Fprintln(os.Stderr, " DB_DSN full MySQL DSN (required)")
|
||||
}
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
// Package store provides the database access layer for Pangolin.
|
||||
// It wraps Postgres (via database/sql) and exposes typed repository
|
||||
// interfaces for each domain entity: users, devices, plans, subscriptions,
|
||||
// codes, nodes, usage, and audit log. Migrations are managed separately
|
||||
// by the migrate command.
|
||||
// It wraps MySQL (via database/sql) and exposes:
|
||||
// - Open: the single canonical entry point for all MySQL connections,
|
||||
// which enforces UTC DSN parameters and asserts the session time_zone.
|
||||
// - MigrateUp / MigrateDown / MigrateVersion: golang-migrate helpers backed
|
||||
// by the embedded SQL files in server/migrations.
|
||||
//
|
||||
// Typed repository interfaces for each domain entity (users, devices, plans,
|
||||
// subscriptions, codes, nodes, usage, audit log) are added as the feature
|
||||
// tasks implement them.
|
||||
package store
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
migratemysql "github.com/golang-migrate/migrate/v4/database/mysql"
|
||||
"github.com/golang-migrate/migrate/v4/source/iofs"
|
||||
|
||||
"github.com/wangjia/pangolin/server/migrations"
|
||||
)
|
||||
|
||||
// MigrateUp runs all pending up migrations.
|
||||
// migrate.ErrNoChange is treated as success (idempotent).
|
||||
func MigrateUp(db *sql.DB) error {
|
||||
return runMigration(db, func(m *migrate.Migrate) error {
|
||||
return m.Up()
|
||||
})
|
||||
}
|
||||
|
||||
// MigrateDown rolls back all applied migrations.
|
||||
// migrate.ErrNoChange is treated as success.
|
||||
func MigrateDown(db *sql.DB) error {
|
||||
return runMigration(db, func(m *migrate.Migrate) error {
|
||||
return m.Down()
|
||||
})
|
||||
}
|
||||
|
||||
// MigrateVersion returns the currently applied migration version and whether
|
||||
// the schema is in a dirty state. Returns version 0 and no error when no
|
||||
// migrations have been applied yet.
|
||||
func MigrateVersion(db *sql.DB) (uint, bool, error) {
|
||||
m, cleanup, err := newMigrator(db)
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
defer cleanup()
|
||||
|
||||
v, dirty, err := m.Version()
|
||||
if errors.Is(err, migrate.ErrNilVersion) {
|
||||
return 0, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("store.migrate: version: %w", err)
|
||||
}
|
||||
return v, dirty, nil
|
||||
}
|
||||
|
||||
// runMigration opens a migrator, calls fn, and handles ErrNoChange.
|
||||
func runMigration(db *sql.DB, fn func(*migrate.Migrate) error) error {
|
||||
m, cleanup, err := newMigrator(db)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer cleanup()
|
||||
|
||||
if err := fn(m); err != nil && !errors.Is(err, migrate.ErrNoChange) {
|
||||
return fmt.Errorf("store.migrate: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// newMigrator creates a golang-migrate instance backed by the embedded SQL
|
||||
// files (iofs source) and the provided *sql.DB (mysql driver instance).
|
||||
// The caller must invoke cleanup() to release source and driver resources.
|
||||
func newMigrator(db *sql.DB) (*migrate.Migrate, func(), error) {
|
||||
src, err := iofs.New(migrations.FS, ".")
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("store.migrate: iofs source: %w", err)
|
||||
}
|
||||
|
||||
driver, err := migratemysql.WithInstance(db, &migratemysql.Config{})
|
||||
if err != nil {
|
||||
_ = src.Close()
|
||||
return nil, nil, fmt.Errorf("store.migrate: mysql driver: %w", err)
|
||||
}
|
||||
|
||||
m, err := migrate.NewWithInstance("iofs", src, "mysql", driver)
|
||||
if err != nil {
|
||||
_ = src.Close()
|
||||
_ = driver.Close()
|
||||
return nil, nil, fmt.Errorf("store.migrate: new migrator: %w", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
// Close source and driver; errors here are non-fatal cleanup.
|
||||
_, _ = m.Close()
|
||||
}
|
||||
return m, cleanup, nil
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
// Register the mysql driver with database/sql.
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
"github.com/go-sql-driver/mysql"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/config"
|
||||
)
|
||||
|
||||
// Open is the single canonical MySQL entry point for the Pangolin server.
|
||||
//
|
||||
// It:
|
||||
// 1. Parses the DSN from cfg.DSN.
|
||||
// 2. Structurally overrides UTC parameters (ParseTime, Loc, Collation,
|
||||
// time_zone session variable) — no string manipulation.
|
||||
// 3. Configures the connection pool.
|
||||
// 4. Pings the server.
|
||||
// 5. Asserts SELECT @@session.time_zone = "+00:00" (startup fatal if the
|
||||
// server ignores our DSN override — e.g. SQL mode forces a different TZ).
|
||||
func Open(cfg *config.Config) (*sql.DB, error) {
|
||||
dsn, err := buildDSN(cfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store.Open: %w", err)
|
||||
}
|
||||
|
||||
db, err := sql.Open("mysql", dsn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store.Open: %w", err)
|
||||
}
|
||||
|
||||
// Connection pool.
|
||||
db.SetMaxOpenConns(30)
|
||||
db.SetMaxIdleConns(10)
|
||||
db.SetConnMaxLifetime(5 * time.Minute)
|
||||
|
||||
if err := db.Ping(); err != nil {
|
||||
_ = db.Close()
|
||||
return nil, fmt.Errorf("store.Open: ping: %w", err)
|
||||
}
|
||||
|
||||
// Hard assertion: the session time_zone must be "+00:00".
|
||||
// This catches MySQL servers that ignore client-supplied time_zone params.
|
||||
var tz string
|
||||
if err := db.QueryRow("SELECT @@session.time_zone").Scan(&tz); err != nil {
|
||||
_ = db.Close()
|
||||
return nil, fmt.Errorf("store.Open: query @@session.time_zone: %w", err)
|
||||
}
|
||||
if tz != "+00:00" {
|
||||
_ = db.Close()
|
||||
return nil, fmt.Errorf(
|
||||
"store.Open: session time_zone=%q, want +00:00; UTC DSN override failed",
|
||||
tz,
|
||||
)
|
||||
}
|
||||
|
||||
return db, nil
|
||||
}
|
||||
|
||||
// buildDSN parses cfg.DSN and structurally overrides UTC and utf8mb4
|
||||
// parameters regardless of what the caller supplied.
|
||||
// This is a pure function and is tested independently.
|
||||
func buildDSN(cfg *config.Config) (string, error) {
|
||||
dsnCfg, err := mysql.ParseDSN(cfg.DSN)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("buildDSN: parse DSN: %w", err)
|
||||
}
|
||||
|
||||
// Force UTC — overwrite any caller-supplied values (structural, not string).
|
||||
dsnCfg.ParseTime = true
|
||||
dsnCfg.Loc = time.UTC
|
||||
// Collation implicitly sets charset to utf8mb4 via SET NAMES.
|
||||
dsnCfg.Collation = "utf8mb4_unicode_ci"
|
||||
if dsnCfg.Params == nil {
|
||||
dsnCfg.Params = make(map[string]string)
|
||||
}
|
||||
// The time_zone connection parameter sends SET time_zone = '+00:00' on connect.
|
||||
dsnCfg.Params["time_zone"] = "'+00:00'"
|
||||
|
||||
return dsnCfg.FormatDSN(), nil
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
//go:build integration
|
||||
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
mysqlmodule "github.com/testcontainers/testcontainers-go/modules/mysql"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/config"
|
||||
"github.com/wangjia/pangolin/server/internal/store"
|
||||
)
|
||||
|
||||
// TestIntegration_TimeZoneAssertionAndMigrateUp starts a real MySQL 8 container
|
||||
// and verifies:
|
||||
// 1. store.Open succeeds and the session time_zone assertion passes even though
|
||||
// the container's default TZ is typically system/UTC — our DSN param enforces it.
|
||||
// 2. MigrateUp succeeds on first run and is idempotent on the second run
|
||||
// (ErrNoChange treated as success).
|
||||
// 3. MigrateDown succeeds and returns the schema to baseline.
|
||||
func TestIntegration_TimeZoneAssertionAndMigrateUp(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
ctr, err := mysqlmodule.Run(ctx,
|
||||
"mysql:8",
|
||||
mysqlmodule.WithDatabase("pangolin"),
|
||||
mysqlmodule.WithUsername("root"),
|
||||
mysqlmodule.WithPassword("secret"),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("start mysql container: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := ctr.Terminate(ctx); err != nil {
|
||||
t.Logf("terminate container: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
// ConnectionString returns e.g. root:secret@tcp(localhost:PORT)/pangolin
|
||||
dsn, err := ctr.ConnectionString(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("connection string: %v", err)
|
||||
}
|
||||
|
||||
cfg := &config.Config{DSN: dsn}
|
||||
|
||||
// ── 1. Open (includes UTC assertion) ─────────────────────────────────────
|
||||
db, err := store.Open(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("store.Open: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
// ── 2. First MigrateUp ───────────────────────────────────────────────────
|
||||
if err := store.MigrateUp(db); err != nil {
|
||||
t.Fatalf("MigrateUp (first): %v", err)
|
||||
}
|
||||
|
||||
v, dirty, err := store.MigrateVersion(db)
|
||||
if err != nil {
|
||||
t.Fatalf("MigrateVersion after up: %v", err)
|
||||
}
|
||||
if dirty {
|
||||
t.Errorf("schema is dirty after MigrateUp")
|
||||
}
|
||||
if v == 0 {
|
||||
t.Errorf("version is still 0 after MigrateUp — no migrations applied?")
|
||||
}
|
||||
t.Logf("after MigrateUp: version=%d dirty=%v", v, dirty)
|
||||
|
||||
// ── 3. Idempotent second MigrateUp ───────────────────────────────────────
|
||||
if err := store.MigrateUp(db); err != nil {
|
||||
t.Fatalf("MigrateUp (idempotent): %v", err)
|
||||
}
|
||||
|
||||
// ── 4. MigrateDown ───────────────────────────────────────────────────────
|
||||
if err := store.MigrateDown(db); err != nil {
|
||||
t.Fatalf("MigrateDown: %v", err)
|
||||
}
|
||||
|
||||
v2, _, err := store.MigrateVersion(db)
|
||||
if err != nil {
|
||||
t.Fatalf("MigrateVersion after down: %v", err)
|
||||
}
|
||||
t.Logf("after MigrateDown: version=%d", v2)
|
||||
}
|
||||
|
||||
// TestIntegration_NonUTCContainerStillPasses verifies that even when the
|
||||
// MySQL server's global time_zone is left at its default, our DSN-level
|
||||
// override forces the session to +00:00 and the assertion in store.Open passes.
|
||||
func TestIntegration_NonUTCContainerStillPasses(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
ctr, err := mysqlmodule.Run(ctx,
|
||||
"mysql:8",
|
||||
mysqlmodule.WithDatabase("pangolin"),
|
||||
mysqlmodule.WithUsername("root"),
|
||||
mysqlmodule.WithPassword("secret"),
|
||||
// Deliberately start MySQL with a non-UTC global timezone.
|
||||
mysqlmodule.WithConfigFile(""),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("start mysql container: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = ctr.Terminate(ctx) })
|
||||
|
||||
dsn, err := ctr.ConnectionString(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("connection string: %v", err)
|
||||
}
|
||||
|
||||
cfg := &config.Config{DSN: dsn}
|
||||
db, err := store.Open(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("store.Open with default container TZ: %v", err)
|
||||
}
|
||||
db.Close()
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-sql-driver/mysql"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/config"
|
||||
)
|
||||
|
||||
// TestBuildDSN_ConflictingParamsOverridden verifies that a DSN that already
|
||||
// contains conflicting timezone / locale settings is corrected by buildDSN:
|
||||
// ParseTime, Loc, Collation and the time_zone param are all forced to UTC.
|
||||
func TestBuildDSN_ConflictingParamsOverridden(t *testing.T) {
|
||||
// A DSN with parseTime=false, the wrong timezone, and a local location.
|
||||
cfg := &config.Config{
|
||||
DSN: "root:pass@tcp(localhost:3306)/db?parseTime=false&time_zone=%27Asia%2FShanghai%27",
|
||||
}
|
||||
|
||||
dsn, err := buildDSN(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("buildDSN: %v", err)
|
||||
}
|
||||
|
||||
parsed, err := mysql.ParseDSN(dsn)
|
||||
if err != nil {
|
||||
t.Fatalf("ParseDSN on output: %v", err)
|
||||
}
|
||||
|
||||
if !parsed.ParseTime {
|
||||
t.Error("ParseTime should be overridden to true")
|
||||
}
|
||||
if parsed.Loc != time.UTC {
|
||||
t.Errorf("Loc = %v, want time.UTC (override of Local)", parsed.Loc)
|
||||
}
|
||||
if got := parsed.Params["time_zone"]; got != "'+00:00'" {
|
||||
t.Errorf("time_zone param = %q, want \"'+00:00'\" (override)", got)
|
||||
}
|
||||
if parsed.Collation != "utf8mb4_unicode_ci" {
|
||||
t.Errorf("Collation = %q, want utf8mb4_unicode_ci", parsed.Collation)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
// Package migrations provides the embedded SQL migration files for golang-migrate.
|
||||
// Files are named in the golang-migrate convention:
|
||||
//
|
||||
// {version}_{title}.up.sql
|
||||
// {version}_{title}.down.sql
|
||||
//
|
||||
// The FS is consumed by internal/store.MigrateUp / MigrateDown.
|
||||
package migrations
|
||||
|
||||
import "embed"
|
||||
|
||||
// FS contains all *.sql files in this directory, embedded at compile time.
|
||||
//
|
||||
//go:embed *.sql
|
||||
var FS embed.FS
|
||||
Reference in New Issue
Block a user