Changeset 0.9.0 (#50)
This commit is contained in:
@@ -1,148 +1,156 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"embed"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
//go:embed migrations/*.sql
|
||||
var migrationsFS embed.FS
|
||||
// schemaVersion tracks the latest applied migration.
|
||||
var schemaVersion string = "none"
|
||||
|
||||
// Migrate checks the database schema state and applies any pending
|
||||
// migrations. This runs at startup before the HTTP server opens.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Ping DB (health check)
|
||||
// 2. Ensure schema_migrations table exists
|
||||
// 3. Load applied versions
|
||||
// 4. Discover embedded SQL files
|
||||
// 5. Apply pending migrations in order
|
||||
//
|
||||
// All migrations run in individual transactions. A failed migration
|
||||
// aborts startup — the backend will not serve traffic with an
|
||||
// inconsistent schema.
|
||||
// SchemaVersion returns the current schema version string.
|
||||
func SchemaVersion() string { return schemaVersion }
|
||||
|
||||
// Migrate runs all pending migrations. It creates the schema_migrations
|
||||
// tracking table if it doesn't exist, then applies each .sql file that
|
||||
// hasn't been applied yet, in order.
|
||||
func Migrate() error {
|
||||
if DB == nil {
|
||||
return fmt.Errorf("database not connected")
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
log.Println("📋 Schema migration check...")
|
||||
|
||||
// ── 1. Health check ─────────────────────────
|
||||
if err := DB.Ping(); err != nil {
|
||||
return fmt.Errorf("database unreachable: %w", err)
|
||||
}
|
||||
|
||||
// ── 2. Ensure tracking table exists ─────────
|
||||
// Ensure tracking table exists
|
||||
_, err := DB.Exec(`
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version VARCHAR(255) PRIMARY KEY,
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at TIMESTAMPTZ DEFAULT NOW()
|
||||
)
|
||||
`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create schema_migrations: %w", err)
|
||||
return fmt.Errorf("create migrations table: %w", err)
|
||||
}
|
||||
|
||||
// ── 3. Load already-applied versions ────────
|
||||
applied := make(map[string]bool)
|
||||
rows, err := DB.Query(`SELECT version FROM schema_migrations`)
|
||||
// Find migration files
|
||||
migrationsDir := findMigrationsDir()
|
||||
if migrationsDir == "" {
|
||||
log.Println("⚠ No migrations directory found — skipping schema migration")
|
||||
return nil
|
||||
}
|
||||
|
||||
entries, err := os.ReadDir(migrationsDir)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read schema_migrations: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var v string
|
||||
if err := rows.Scan(&v); err != nil {
|
||||
return fmt.Errorf("scan version: %w", err)
|
||||
}
|
||||
applied[v] = true
|
||||
}
|
||||
|
||||
// ── 4. Discover embedded migration files ────
|
||||
entries, err := migrationsFS.ReadDir("migrations")
|
||||
if err != nil {
|
||||
return fmt.Errorf("read embedded migrations: %w", err)
|
||||
return fmt.Errorf("read migrations dir: %w", err)
|
||||
}
|
||||
|
||||
// Collect and sort .sql files
|
||||
var files []string
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
|
||||
files = append(files, e.Name())
|
||||
}
|
||||
}
|
||||
sort.Strings(files) // lexicographic = version order (001_, 002_, ...)
|
||||
sort.Strings(files)
|
||||
|
||||
// ── 5. Apply pending ────────────────────────
|
||||
pending := 0
|
||||
skipped := 0
|
||||
for _, name := range files {
|
||||
if applied[name] {
|
||||
skipped++
|
||||
if len(files) == 0 {
|
||||
log.Println(" No migration files found")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Compat: rename old numeric-only version entries to full filenames.
|
||||
// Earlier extractVersion used regex ^(\d+), recording "001" instead of
|
||||
// "001_v09_schema.sql". Fix them in-place so CI's db-migrate.sh matches.
|
||||
for _, file := range files {
|
||||
prefix := strings.SplitN(file, "_", 2)[0] // "001"
|
||||
DB.Exec("UPDATE schema_migrations SET version = $1 WHERE version = $2 AND version != $1", file, prefix)
|
||||
}
|
||||
|
||||
// Apply pending migrations
|
||||
applied := 0
|
||||
for _, file := range files {
|
||||
version := extractVersion(file)
|
||||
if version == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
content, err := migrationsFS.ReadFile("migrations/" + name)
|
||||
// Check if already applied
|
||||
var exists bool
|
||||
DB.QueryRow("SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)", version).Scan(&exists)
|
||||
if exists {
|
||||
schemaVersion = version
|
||||
continue
|
||||
}
|
||||
|
||||
// Read and execute
|
||||
path := filepath.Join(migrationsDir, file)
|
||||
sql, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read migration %s: %w", name, err)
|
||||
return fmt.Errorf("read %s: %w", file, err)
|
||||
}
|
||||
|
||||
log.Printf(" ▶ applying: %s", name)
|
||||
|
||||
tx, err := DB.Begin()
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin tx for %s: %w", name, err)
|
||||
log.Printf(" Applying migration %s...", file)
|
||||
if _, err := DB.Exec(string(sql)); err != nil {
|
||||
return fmt.Errorf("apply %s: %w", file, err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(string(content)); err != nil {
|
||||
tx.Rollback()
|
||||
return fmt.Errorf("migration %s failed: %w", name, err)
|
||||
// Record
|
||||
if _, err := DB.Exec("INSERT INTO schema_migrations (version) VALUES ($1)", version); err != nil {
|
||||
return fmt.Errorf("record %s: %w", file, err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(
|
||||
`INSERT INTO schema_migrations (version) VALUES ($1)`, name,
|
||||
); err != nil {
|
||||
tx.Rollback()
|
||||
return fmt.Errorf("record migration %s: %w", name, err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit migration %s: %w", name, err)
|
||||
}
|
||||
|
||||
log.Printf(" ✓ %s applied", name)
|
||||
pending++
|
||||
schemaVersion = version
|
||||
applied++
|
||||
}
|
||||
|
||||
elapsed := time.Since(start).Round(time.Millisecond)
|
||||
if pending > 0 {
|
||||
log.Printf("✅ Migrations complete: %d applied, %d skipped (%s)", pending, skipped, elapsed)
|
||||
if applied > 0 {
|
||||
log.Printf(" ✅ Applied %d migration(s), schema at %s", applied, schemaVersion)
|
||||
} else {
|
||||
log.Printf("✅ Schema up to date (%d migrations, %s)", skipped, elapsed)
|
||||
log.Printf(" Schema up to date at %s", schemaVersion)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// SchemaVersion returns the most recently applied migration version,
|
||||
// or "" if no migrations have been applied.
|
||||
func SchemaVersion() string {
|
||||
if DB == nil {
|
||||
// extractVersion returns the filename as the version key if it's a valid
|
||||
// migration file (starts with digit, ends with .sql).
|
||||
// "001_v09_schema.sql" → "001_v09_schema.sql" (matches db-migrate.sh convention)
|
||||
func extractVersion(filename string) string {
|
||||
// Use full filename as version to match db-migrate.sh convention
|
||||
if !strings.HasSuffix(filename, ".sql") {
|
||||
return ""
|
||||
}
|
||||
var version sql.NullString
|
||||
err := DB.QueryRow(`
|
||||
SELECT version FROM schema_migrations
|
||||
ORDER BY version DESC LIMIT 1
|
||||
`).Scan(&version)
|
||||
if err != nil || !version.Valid {
|
||||
// Must start with a digit (e.g., 001_v09_schema.sql)
|
||||
if len(filename) == 0 || filename[0] < '0' || filename[0] > '9' {
|
||||
return ""
|
||||
}
|
||||
return version.String
|
||||
return filename
|
||||
}
|
||||
|
||||
// findMigrationsDir locates the migrations directory.
|
||||
// Checks relative to the binary, then relative to the source file.
|
||||
func findMigrationsDir() string {
|
||||
candidates := []string{
|
||||
"database/migrations",
|
||||
"server/database/migrations",
|
||||
"../database/migrations",
|
||||
}
|
||||
|
||||
// Also check relative to this source file (for tests)
|
||||
_, thisFile, _, ok := runtime.Caller(0)
|
||||
if ok {
|
||||
dir := filepath.Dir(thisFile)
|
||||
candidates = append(candidates, filepath.Join(dir, "migrations"))
|
||||
}
|
||||
|
||||
for _, c := range candidates {
|
||||
if info, err := os.Stat(c); err == nil && info.IsDir() {
|
||||
return c
|
||||
}
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user