ballet-server/backend/internal/database/migrate.go

package database

import (
	"context"
	_ "embed"
	"fmt"

	"github.com/jackc/pgx/v5/pgxpool"
)

//go:embed 001_practice.sql
var practiceSQL string

//go:embed 002_project_icons.sql
var projectIconsSQL string

//go:embed 003_project_order.sql
var projectOrderSQL string

// Migrate applies versioned migrations atomically. A transaction-scoped lock keeps
// concurrent service starts from applying the same version twice.
func Migrate(ctx context.Context, pool *pgxpool.Pool) error {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)
	if _, err = tx.Exec(ctx, "SELECT pg_advisory_xact_lock(684127930)"); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, "CREATE TABLE IF NOT EXISTS schema_migrations (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())"); err != nil {
		return err
	}
	for i, sql := range []string{practiceSQL, projectIconsSQL, projectOrderSQL} {
		version := i + 1
		var applied bool
		if err = tx.QueryRow(ctx, "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version=$1)", version).Scan(&applied); err != nil {
			return err
		}
		if !applied {
			if _, err = tx.Exec(ctx, sql); err != nil {
				return fmt.Errorf("migration %d: %w", version, err)
			}
			if _, err = tx.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES ($1)", version); err != nil {
				return err
			}
		}
	}
	return tx.Commit(ctx)
}