authkit/store_migrate.go

package authkit

import (
	"context"
	"database/sql"
	"embed"
	"io/fs"
	"log"
	"sort"
	"strings"

	"git.juancwu.dev/juancwu/errx"
)

//go:embed migrations/*.sql
var migrationsFS embed.FS

// advisoryLockKey is the ASCII bytes of "authkit" packed into an int64.
// Stable across rollouts and unlikely to clash with caller advisory locks.
const advisoryLockKey int64 = 0x617574686b6974

// Migrate applies every embedded migration not yet recorded in the
// schema-migrations table. Safe to call repeatedly and concurrently across
// processes; the advisory lock serialises rollouts. Each migration owns its
// own BEGIN/COMMIT.
//
// Embedded migrations hard-code the default authkit_* names. If the consumer
// has overridden any table name, Migrate is a no-op and the consumer is
// responsible for managing DDL out-of-band.
func Migrate(ctx context.Context, db *sql.DB, schema Schema) error {
	const op = "authkit.Migrate"
	if db == nil {
		return errx.New(op, "db is required")
	}
	if err := schema.Validate(); err != nil {
		return errx.Wrap(op, err)
	}

	if !schema.isDefault() {
		// Custom-named schemas: consumer owns DDL. The verifier still runs
		// against the configured names (with default-name fallback) to
		// confirm the tables exist and match the expected layout.
		return nil
	}

	conn, err := db.Conn(ctx)
	if err != nil {
		return errx.Wrap(op, err)
	}
	defer conn.Close()

	if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_lock($1)", advisoryLockKey); err != nil {
		return errx.Wrap(op, err)
	}
	defer func() {
		if _, err := conn.ExecContext(context.Background(),
			"SELECT pg_advisory_unlock($1)", advisoryLockKey); err != nil {
			log.Printf("authkit: pg_advisory_unlock failed: %v", err)
		}
	}()

	q := buildQueries(schema.Tables)
	if _, err := conn.ExecContext(ctx, q.createMigrationsTable); err != nil {
		return errx.Wrap(op, err)
	}

	applied, err := loadAppliedVersions(ctx, conn, q.selectAppliedVersions)
	if err != nil {
		return errx.Wrap(op, err)
	}

	migs, err := fs.Sub(migrationsFS, "migrations")
	if err != nil {
		return errx.Wrap(op, err)
	}
	files, err := fs.ReadDir(migs, ".")
	if err != nil {
		return errx.Wrap(op, err)
	}
	names := make([]string, 0, len(files))
	for _, f := range files {
		if !f.IsDir() && strings.HasSuffix(f.Name(), ".sql") {
			names = append(names, f.Name())
		}
	}
	sort.Strings(names)

	for _, name := range names {
		version := strings.TrimSuffix(name, ".sql")
		if _, ok := applied[version]; ok {
			continue
		}
		body, err := fs.ReadFile(migs, name)
		if err != nil {
			return errx.Wrapf(op, err, "read %s", name)
		}
		if _, err := conn.ExecContext(ctx, string(body)); err != nil {
			return errx.Wrapf(op, err, "apply %s", version)
		}
	}
	return nil
}

func loadAppliedVersions(ctx context.Context, conn *sql.Conn, q string) (map[string]struct{}, error) {
	rows, err := conn.QueryContext(ctx, q)
	if err != nil {
		if isMissingTable(err) {
			return map[string]struct{}{}, nil
		}
		return nil, err
	}
	defer rows.Close()
	out := make(map[string]struct{})
	for rows.Next() {
		var v string
		if err := rows.Scan(&v); err != nil {
			return nil, err
		}
		out[v] = struct{}{}
	}
	return out, rows.Err()
}