149 lines
3.7 KiB
Go
149 lines
3.7 KiB
Go
package database
|
|
|
|
import (
|
|
"database/sql"
|
|
"embed"
|
|
"fmt"
|
|
"log"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
//go:embed migrations/*.sql
|
|
var migrationsFS embed.FS
|
|
|
|
// 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.
|
|
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 ─────────
|
|
_, err := DB.Exec(`
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version VARCHAR(255) PRIMARY KEY,
|
|
applied_at TIMESTAMPTZ DEFAULT NOW()
|
|
)
|
|
`)
|
|
if err != nil {
|
|
return fmt.Errorf("create schema_migrations: %w", err)
|
|
}
|
|
|
|
// ── 3. Load already-applied versions ────────
|
|
applied := make(map[string]bool)
|
|
rows, err := DB.Query(`SELECT version FROM schema_migrations`)
|
|
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)
|
|
}
|
|
|
|
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_, ...)
|
|
|
|
// ── 5. Apply pending ────────────────────────
|
|
pending := 0
|
|
skipped := 0
|
|
for _, name := range files {
|
|
if applied[name] {
|
|
skipped++
|
|
continue
|
|
}
|
|
|
|
content, err := migrationsFS.ReadFile("migrations/" + name)
|
|
if err != nil {
|
|
return fmt.Errorf("read migration %s: %w", name, err)
|
|
}
|
|
|
|
log.Printf(" ▶ applying: %s", name)
|
|
|
|
tx, err := DB.Begin()
|
|
if err != nil {
|
|
return fmt.Errorf("begin tx for %s: %w", name, err)
|
|
}
|
|
|
|
if _, err := tx.Exec(string(content)); err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("migration %s failed: %w", name, 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++
|
|
}
|
|
|
|
elapsed := time.Since(start).Round(time.Millisecond)
|
|
if pending > 0 {
|
|
log.Printf("✅ Migrations complete: %d applied, %d skipped (%s)", pending, skipped, elapsed)
|
|
} else {
|
|
log.Printf("✅ Schema up to date (%d migrations, %s)", skipped, elapsed)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// SchemaVersion returns the most recently applied migration version,
|
|
// or "" if no migrations have been applied.
|
|
func SchemaVersion() string {
|
|
if DB == nil {
|
|
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 {
|
|
return ""
|
|
}
|
|
return version.String
|
|
}
|