package database import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "fmt" "io/fs" "log/slog" "sort" "strconv" "strings" "calllinesystem/server/migrations" ) const migrationLockID int64 = 733081337591910 func Migrate(ctx context.Context, db *sql.DB, logger *slog.Logger) error { conn, err := db.Conn(ctx) if err != nil { return fmt.Errorf("acquire migration connection: %w", err) } defer conn.Close() if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock($1)`, migrationLockID); err != nil { return fmt.Errorf("lock migrations: %w", err) } defer func() { _, _ = conn.ExecContext(context.Background(), `SELECT pg_advisory_unlock($1)`, migrationLockID) }() if _, err := conn.ExecContext(ctx, ` CREATE TABLE IF NOT EXISTS schema_migrations ( version bigint PRIMARY KEY, name text NOT NULL, checksum char(64) NOT NULL, applied_at timestamptz NOT NULL DEFAULT now() )`); err != nil { return fmt.Errorf("create schema_migrations: %w", err) } applied, err := readApplied(ctx, conn) if err != nil { return err } entries, err := fs.ReadDir(migrations.Files, ".") if err != nil { return fmt.Errorf("read embedded migrations: %w", err) } var names []string for _, entry := range entries { if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".up.sql") { names = append(names, entry.Name()) } } sort.Strings(names) for _, name := range names { version, err := migrationVersion(name) if err != nil { return err } body, err := migrations.Files.ReadFile(name) if err != nil { return fmt.Errorf("read migration %s: %w", name, err) } sumBytes := sha256.Sum256(body) checksum := hex.EncodeToString(sumBytes[:]) if existing, ok := applied[version]; ok { if existing != checksum { return fmt.Errorf("migration %d checksum changed", version) } continue } tx, err := conn.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin migration %s: %w", name, err) } if _, err = tx.ExecContext(ctx, string(body)); err == nil { _, err = tx.ExecContext(ctx, `INSERT INTO schema_migrations (version, name, checksum) VALUES ($1, $2, $3)`, version, name, checksum, ) } if err != nil { _ = tx.Rollback() return fmt.Errorf("apply migration %s: %w", name, err) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit migration %s: %w", name, err) } logger.Info("database migration applied", "version", version, "name", name) } return nil } // LatestVersion returns the highest embedded migration version. It is used by // readiness checks so an API Pod cannot report ready against an older schema. func LatestVersion() (int64, error) { entries, err := fs.ReadDir(migrations.Files, ".") if err != nil { return 0, fmt.Errorf("read embedded migrations: %w", err) } var latest int64 for _, entry := range entries { if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".up.sql") { continue } version, err := migrationVersion(entry.Name()) if err != nil { return 0, err } if version > latest { latest = version } } if latest == 0 { return 0, fmt.Errorf("no embedded migrations found") } return latest, nil } // SchemaReady verifies that all embedded migrations have been applied. func SchemaReady(ctx context.Context, db *sql.DB) error { latest, err := LatestVersion() if err != nil { return err } var applied int64 if err := db.QueryRowContext(ctx, `SELECT COALESCE(MAX(version), 0) FROM schema_migrations`).Scan(&applied); err != nil { return fmt.Errorf("read applied migration version: %w", err) } if applied < latest { return fmt.Errorf("database schema version %d is behind embedded version %d", applied, latest) } return nil } func readApplied(ctx context.Context, conn *sql.Conn) (map[int64]string, error) { rows, err := conn.QueryContext(ctx, `SELECT version, checksum FROM schema_migrations ORDER BY version`) if err != nil { return nil, fmt.Errorf("read schema migrations: %w", err) } defer rows.Close() applied := make(map[int64]string) for rows.Next() { var version int64 var checksum string if err := rows.Scan(&version, &checksum); err != nil { return nil, fmt.Errorf("scan schema migration: %w", err) } applied[version] = checksum } return applied, rows.Err() } func migrationVersion(name string) (int64, error) { prefix, _, ok := strings.Cut(name, "_") if !ok { return 0, fmt.Errorf("invalid migration filename %q", name) } version, err := strconv.ParseInt(prefix, 10, 64) if err != nil || version <= 0 { return 0, fmt.Errorf("invalid migration version in %q", name) } return version, nil }