wangjia/pangolin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

feat(server/db): 数据层多库支持(1/4)— 连接分派 + 双方言迁移管线

Browse Source 
- config 增 DB_DRIVER(mysql 默认 | sqlite);DSN 对 sqlite 为文件路径
- db.OpenDriver 按驱动分派:sqlite 用 modernc(纯 Go 免 CGO)+ WAL/
  busy_timeout/foreign_keys/_txlock=immediate;mysql 路径不变
- store.Open 分派;mysql 保留 UTC/collation 断言,sqlite 跳过
- 迁移拆 migrations/{mysql,sqlite}/ 双套,embed 双 FS,migrate 按驱动选源
  与 golang-migrate 驱动;修复 m.Close() 误关调用方 *sql.DB 的坑
- cmd/migrate 串入 DB_DRIVER;集成测试 MigrateUp 签名更新
- 新增 SQLite 时间往返 smoke 测试与端到端迁移测试(免 docker)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-18 00:01:03 +08:00
parent 9a028ab907
commit f3471ae139
63 changed files with 583 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/store/migrate.go
+47 -22
View File
@@ -4,26 +4,29 @@ import (
"database/sql"
"errors"
"fmt"
"io/fs"
"github.com/golang-migrate/migrate/v4"
migratedb "github.com/golang-migrate/migrate/v4/database"
migratemysql "github.com/golang-migrate/migrate/v4/database/mysql"
migratesqlite "github.com/golang-migrate/migrate/v4/database/sqlite"
"github.com/golang-migrate/migrate/v4/source/iofs"
"github.com/wangjia/pangolin/server/internal/db"
"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 {
// MigrateUp runs all pending up migrations for the given driver
// ("mysql" | "sqlite"). migrate.ErrNoChange is treated as success (idempotent).
func MigrateUp(database *sql.DB, driver string) error {
return runMigration(database, driver, 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 {
// MigrateDown rolls back all applied migrations. ErrNoChange = success.
func MigrateDown(database *sql.DB, driver string) error {
return runMigration(database, driver, func(m *migrate.Migrate) error {
return m.Down()
})
}
@@ -31,8 +34,8 @@ func MigrateDown(db *sql.DB) error {
// 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)
func MigrateVersion(database *sql.DB, driver string) (uint, bool, error) {
m, cleanup, err := newMigrator(database, driver)
if err != nil {
return 0, false, err
}
@@ -49,8 +52,8 @@ func MigrateVersion(db *sql.DB) (uint, bool, error) {
}
// 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)
func runMigration(database *sql.DB, driver string, fn func(*migrate.Migrate) error) error {
m, cleanup, err := newMigrator(database, driver)
if err != nil {
return err
}
@@ -63,30 +66,52 @@ func runMigration(db *sql.DB, fn func(*migrate.Migrate) error) error {
}
// 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, ".")
// files for the chosen dialect and the provided *sql.DB. The caller must invoke
// cleanup() to release source and driver resources.
func newMigrator(database *sql.DB, driver string) (*migrate.Migrate, func(), error) {
embedFS, sub, dbName := migrationSource(driver)
src, err := iofs.New(embedFS, sub)
if err != nil {
return nil, nil, fmt.Errorf("store.migrate: iofs source: %w", err)
}
driver, err := migratemysql.WithInstance(db, &migratemysql.Config{})
mdriver, err := newMigrateDriver(driver, database)
if err != nil {
_ = src.Close()
return nil, nil, fmt.Errorf("store.migrate: mysql driver: %w", err)
return nil, nil, fmt.Errorf("store.migrate: %s driver: %w", dbName, err)
}
m, err := migrate.NewWithInstance("iofs", src, "mysql", driver)
m, err := migrate.NewWithInstance("iofs", src, dbName, mdriver)
if err != nil {
_ = src.Close()
_ = driver.Close()
_ = mdriver.Close()
return nil, nil, fmt.Errorf("store.migrate: new migrator: %w", err)
}
// Close only the embedded source, NOT the database driver: m.Close() would
// close the caller-owned *sql.DB (the WithInstance driver closes the handle
// it was given — fatal for a shared/in-memory connection the caller reuses).
// The caller owns the *sql.DB and closes it when done.
cleanup := func() {
// Close source and driver; errors here are non-fatal cleanup.
_, _ = m.Close()
_ = src.Close()
}
return m, cleanup, nil
}
// migrationSource returns the embedded FS, its subdir, and the golang-migrate
// database name for the given driver.
func migrationSource(driver string) (fs.FS, string, string) {
if db.Normalize(driver) == "sqlite" {
return migrations.SQLiteFS, "sqlite", "sqlite"
}
return migrations.MySQLFS, "mysql", "mysql"
}
// newMigrateDriver builds the golang-migrate database driver for the dialect.
func newMigrateDriver(driver string, database *sql.DB) (migratedb.Driver, error) {
if db.Normalize(driver) == "sqlite" {
return migratesqlite.WithInstance(database, &migratesqlite.Config{})
}
return migratemysql.WithInstance(database, &migratemysql.Config{})
}
server/internal/store/mysql.go
+21 -26
View File
@@ -5,60 +5,55 @@ import (
"fmt"
"time"
// Register the mysql driver with database/sql.
// mysql.ParseDSN is used to structurally rewrite the DSN for UTC.
"github.com/go-sql-driver/mysql"
_ "github.com/go-sql-driver/mysql"
"github.com/wangjia/pangolin/server/internal/config"
"github.com/wangjia/pangolin/server/internal/db"
)
// Open is the single canonical MySQL entry point for the Pangolin server.
// Open is the canonical database entry point for the Pangolin server. It
// dispatches on cfg.Driver:
//
// 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).
// - sqlite: opens the file/in-memory DB (UTC is native; no session assertion).
// - mysql (default): structurally overrides the DSN for UTC (ParseTime, Loc,
// Collation, time_zone), opens the pool, and asserts
// SELECT @@session.time_zone = "+00:00" (startup-fatal if the server ignores
// our UTC override — e.g. SQL mode forces a different TZ).
//
// The actual connection/pool/ping lives in internal/db; this layer adds only the
// MySQL-specific UTC rigor.
func Open(cfg *config.Config) (*sql.DB, error) {
if db.Normalize(cfg.Driver) == "sqlite" {
return db.OpenDriver("sqlite", cfg.DSN)
}
dsn, err := buildDSN(cfg)
if err != nil {
return nil, fmt.Errorf("store.Open: %w", err)
}
db, err := sql.Open("mysql", dsn)
database, err := db.OpenDriver("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()
if err := database.QueryRow("SELECT @@session.time_zone").Scan(&tz); err != nil {
_ = database.Close()
return nil, fmt.Errorf("store.Open: query @@session.time_zone: %w", err)
}
if tz != "+00:00" {
_ = db.Close()
_ = database.Close()
return nil, fmt.Errorf(
"store.Open: session time_zone=%q, want +00:00; UTC DSN override failed",
tz,
)
}
return db, nil
return database, nil
}
// buildDSN parses cfg.DSN and structurally overrides UTC and utf8mb4
server/internal/store/mysql_integration_test.go
+5 -5
View File
@@ -53,11 +53,11 @@ func TestIntegration_TimeZoneAssertionAndMigrateUp(t *testing.T) {
defer db.Close()
// ── 2. First MigrateUp ───────────────────────────────────────────────────
if err := store.MigrateUp(db); err != nil {
if err := store.MigrateUp(db, "mysql"); err != nil {
t.Fatalf("MigrateUp (first): %v", err)
}
v, dirty, err := store.MigrateVersion(db)
v, dirty, err := store.MigrateVersion(db, "mysql")
if err != nil {
t.Fatalf("MigrateVersion after up: %v", err)
}
@@ -70,16 +70,16 @@ func TestIntegration_TimeZoneAssertionAndMigrateUp(t *testing.T) {
t.Logf("after MigrateUp: version=%d dirty=%v", v, dirty)
// ── 3. Idempotent second MigrateUp ───────────────────────────────────────
if err := store.MigrateUp(db); err != nil {
if err := store.MigrateUp(db, "mysql"); err != nil {
t.Fatalf("MigrateUp (idempotent): %v", err)
}
// ── 4. MigrateDown ───────────────────────────────────────────────────────
if err := store.MigrateDown(db); err != nil {
if err := store.MigrateDown(db, "mysql"); err != nil {
t.Fatalf("MigrateDown: %v", err)
}
v2, _, err := store.MigrateVersion(db)
v2, _, err := store.MigrateVersion(db, "mysql")
if err != nil {
t.Fatalf("MigrateVersion after down: %v", err)
}
server/internal/store/sqlite_migrate_test.go
+93
View File
@@ -0,0 +1,93 @@
package store_test
import (
"testing"
"github.com/wangjia/pangolin/server/internal/config"
"github.com/wangjia/pangolin/server/internal/store"
)
// TestSQLiteMigrateUpDown verifies the SQLite migration set applies cleanly,
// is idempotent, seeds correctly, and rolls back — no container required, so it
// runs in normal CI (unlike the MySQL integration test behind //go:build integration).
func TestSQLiteMigrateUpDown(t *testing.T) {
cfg := &config.Config{Driver: "sqlite", DSN: ":memory:"}
db, err := store.Open(cfg)
if err != nil {
t.Fatalf("store.Open(sqlite): %v", err)
}
defer db.Close()
// 1. MigrateUp.
if err := store.MigrateUp(db, "sqlite"); err != nil {
t.Fatalf("MigrateUp: %v", err)
}
v, dirty, err := store.MigrateVersion(db, "sqlite")
if err != nil {
t.Fatalf("MigrateVersion: %v", err)
}
if dirty {
t.Fatalf("schema dirty after MigrateUp")
}
if v != 13 {
t.Errorf("version = %d, want 13", v)
}
// 2. Core tables exist.
for _, tbl := range []string{
"users", "devices", "plans", "subscriptions", "code_batches", "codes",
"usage_daily", "audit_log", "providers", "nodes", "node_events",
"directory_version", "provision_idempotency", "replacements", "admins",
"connect_credentials",
} {
var name string
err := db.QueryRow(
`SELECT name FROM sqlite_master WHERE type='table' AND name=?`, tbl,
).Scan(&name)
if err != nil {
t.Errorf("table %q missing: %v", tbl, err)
}
}
// 3. Seed: 3 plans + directory_version singleton.
var plans int
if err := db.QueryRow(`SELECT COUNT(*) FROM plans`).Scan(&plans); err != nil {
t.Fatalf("count plans: %v", err)
}
if plans != 3 {
t.Errorf("plans seeded = %d, want 3", plans)
}
var dv int
if err := db.QueryRow(`SELECT version FROM directory_version WHERE id=1`).Scan(&dv); err != nil {
t.Errorf("directory_version singleton missing: %v", err)
}
// 4. Added columns from later migrations are present (000011 / 000013).
if _, err := db.Exec(`SELECT reality_prk, reality_short_id, sub_token, totp_enabled FROM nodes
LEFT JOIN users ON 0=1 LIMIT 0`); err != nil {
// Separate queries — the join above is just a cheap column-existence probe.
if _, e := db.Exec(`SELECT reality_prk, reality_short_id FROM nodes LIMIT 0`); e != nil {
t.Errorf("nodes reality cols missing: %v", e)
}
if _, e := db.Exec(`SELECT sub_token, totp_secret_enc, totp_enabled FROM users LIMIT 0`); e != nil {
t.Errorf("users totp cols missing: %v", e)
}
}
// 5. Idempotent second MigrateUp.
if err := store.MigrateUp(db, "sqlite"); err != nil {
t.Fatalf("MigrateUp (idempotent): %v", err)
}
// 6. MigrateDown returns to baseline.
if err := store.MigrateDown(db, "sqlite"); err != nil {
t.Fatalf("MigrateDown: %v", err)
}
v2, _, err := store.MigrateVersion(db, "sqlite")
if err != nil {
t.Fatalf("MigrateVersion after down: %v", err)
}
if v2 != 0 {
t.Errorf("version after down = %d, want 0", v2)
}
}