Initial commit

server/internal/database/migrate.go

@@ -0,0 +1,170 @@
+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
+}