nexarch/common/migrate/migrate.go

// Package migrate implements versioned, revertible SQL migrations (no ORM).
// Jedes Modul benennt seine Fortschrittstabelle selbst (New), damit sich
// Module in derselben Datenbank nicht gegenseitig den Fortschritt
// ueberschreiben (FDN-13).
package migrate

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strings"

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

// Migration ist eine einzelne versionierte Migrationsdatei.
type Migration struct {
	Version string // Dateiname ohne .up.sql/.down.sql, z.B. "0001_documents"
	UpSQL   string
	DownSQL string
}

// Load liest alle *.up.sql/*.down.sql-Paare aus dir, sortiert nach
// Dateiname (Akzeptanzkriterium: Migrationen sind versioniert).
func Load(dir string) ([]Migration, error) {
	entries, err := os.ReadDir(dir)
	if err != nil {
		return nil, fmt.Errorf("migrationsverzeichnis %q lesen: %w", dir, err)
	}
	var versions []string
	for _, e := range entries {
		if e.IsDir() || !strings.HasSuffix(e.Name(), ".up.sql") {
			continue
		}
		versions = append(versions, strings.TrimSuffix(e.Name(), ".up.sql"))
	}
	sort.Strings(versions)

	migrations := make([]Migration, 0, len(versions))
	for _, v := range versions {
		up, err := os.ReadFile(filepath.Join(dir, v+".up.sql"))
		if err != nil {
			return nil, fmt.Errorf("migration %q: up.sql lesen: %w", v, err)
		}
		down, err := os.ReadFile(filepath.Join(dir, v+".down.sql"))
		if err != nil {
			return nil, fmt.Errorf("migration %q: down.sql lesen (jede Migration braucht ein Rollback): %w", v, err)
		}
		migrations = append(migrations, Migration{Version: v, UpSQL: string(up), DownSQL: string(down)})
	}
	return migrations, nil
}

var tableNamePattern = regexp.MustCompile(`^[a-z_][a-z0-9_]{0,62}$`)

// Migrator fuehrt Migrationen gegen eine modulspezifisch benannte Tabelle aus.
type Migrator struct {
	table string
}

// New liefert einen Migrator fuer die Fortschrittstabelle table (z. B.
// "dms_schema_migrations"). Der Name geht unquotiert in SQL und muss daher
// ^[a-z_][a-z0-9_]{0,62}$ entsprechen.
func New(table string) (*Migrator, error) {
	if !tableNamePattern.MatchString(table) {
		return nil, fmt.Errorf("migrate: ungueltiger tabellenname %q", table)
	}
	return &Migrator{table: table}, nil
}

// MustNew wie New, panikt bei ungueltigem Namen (fuer Paketvariablen).
func MustNew(table string) *Migrator {
	mg, err := New(table)
	if err != nil {
		panic(err)
	}
	return mg
}

// Table liefert den Namen der Fortschrittstabelle.
func (mg *Migrator) Table() string { return mg.table }

const legacyTrackingTable = "schema_migrations"

// ensureTrackingTable legt mg.table an. Beim erstmaligen Anlegen wird
// ein vorhandener Bestand aus der frueheren, unqualifizierten
// schema_migrations uebernommen (nur Versionen, die dieses Modul kennt, und
// nur wenn dessen Basisversion dort steht — eine schema_migrations eines
// anderen Moduls oder eines anderen Schemas wird weder gelesen noch
// veraendert). Die alte Tabelle
// bleibt unangetastet.
func (mg *Migrator) ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) error {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return fmt.Errorf("%s anlegen: %w", mg.table, err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	// Serialisiert gleichzeitige Erstanlage/Uebernahme (mehrere App-Instanzen).
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext($1))`, mg.table); err != nil {
		return fmt.Errorf("%s anlegen: %w", mg.table, err)
	}

	var existed bool
	if err := tx.QueryRow(ctx, `SELECT to_regclass(current_schema() || '.' || $1::text) IS NOT NULL`, mg.table).Scan(&existed); err != nil {
		return fmt.Errorf("%s anlegen: %w", mg.table, err)
	}
	if existed {
		return tx.Commit(ctx)
	}

	if _, err := tx.Exec(ctx, `
		CREATE TABLE `+mg.table+` (
			version    TEXT PRIMARY KEY,
			applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
		)
	`); err != nil {
		return fmt.Errorf("%s anlegen: %w", mg.table, err)
	}

	if len(migrations) > 0 {
		var legacy bool
		if err := tx.QueryRow(ctx, `SELECT to_regclass(current_schema() || '.' || $1::text) IS NOT NULL`, legacyTrackingTable).Scan(&legacy); err != nil {
			return fmt.Errorf("%s pruefen: %w", legacyTrackingTable, err)
		}
		if legacy {
			versions := make([]string, len(migrations))
			for i, m := range migrations {
				versions[i] = m.Version
			}
			var hasBase bool
			if err := tx.QueryRow(ctx,
				`SELECT EXISTS (SELECT 1 FROM `+legacyTrackingTable+` WHERE version = $1)`, versions[0],
			).Scan(&hasBase); err != nil {
				return fmt.Errorf("%s lesen: %w", legacyTrackingTable, err)
			}
			if hasBase {
				if _, err := tx.Exec(ctx, `
					INSERT INTO `+mg.table+` (version, applied_at)
					SELECT version, applied_at FROM `+legacyTrackingTable+` WHERE version = ANY($1)
				`, versions); err != nil {
					return fmt.Errorf("bestand aus %s uebernehmen: %w", legacyTrackingTable, err)
				}
			}
		}
	}
	return tx.Commit(ctx)
}

func (mg *Migrator) appliedVersions(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) {
	rows, err := pool.Query(ctx, `SELECT version FROM `+mg.table)
	if err != nil {
		return nil, fmt.Errorf("angewendete migrationen lesen: %w", err)
	}
	defer rows.Close()

	applied := map[string]bool{}
	for rows.Next() {
		var v string
		if err := rows.Scan(&v); err != nil {
			return nil, fmt.Errorf("migrationsversion lesen: %w", err)
		}
		applied[v] = true
	}
	return applied, rows.Err()
}

// Up wendet alle noch nicht angewendeten Migrationen in Reihenfolge an
// (Akzeptanzkriterium 2: vorwaerts ausfuehrbar) — bereits angewendete
// werden uebersprungen, damit Up auf einer leeren UND auf einer bestehenden
// DB funktioniert (Pruefung 1).
func (mg *Migrator) Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applied []string, err error) {
	if err := mg.ensureTrackingTable(ctx, pool, migrations); err != nil {
		return nil, err
	}
	already, err := mg.appliedVersions(ctx, pool)
	if err != nil {
		return nil, err
	}

	for _, m := range migrations {
		if already[m.Version] {
			continue
		}
		tx, err := pool.Begin(ctx)
		if err != nil {
			return applied, fmt.Errorf("transaktion fuer %q starten: %w", m.Version, err)
		}
		if _, err := tx.Exec(ctx, m.UpSQL); err != nil {
			_ = tx.Rollback(ctx)
			return applied, fmt.Errorf("migration %q anwenden: %w", m.Version, err)
		}
		if _, err := tx.Exec(ctx, `INSERT INTO `+mg.table+` (version) VALUES ($1)`, m.Version); err != nil {
			_ = tx.Rollback(ctx)
			return applied, fmt.Errorf("migration %q als angewendet markieren: %w", m.Version, err)
		}
		if err := tx.Commit(ctx); err != nil {
			return applied, fmt.Errorf("migration %q committen: %w", m.Version, err)
		}
		applied = append(applied, m.Version)
	}
	return applied, nil
}

// DownOne macht die zuletzt angewendete Migration rueckgaengig
// (Akzeptanzkriterium 2: rueckwaerts ausfuehrbar) und liefert deren Version,
// oder "" falls keine Migration angewendet war.
func (mg *Migrator) DownOne(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (version string, err error) {
	if err := mg.ensureTrackingTable(ctx, pool, migrations); err != nil {
		return "", err
	}
	already, err := mg.appliedVersions(ctx, pool)
	if err != nil {
		return "", err
	}

	var last *Migration
	for i := len(migrations) - 1; i >= 0; i-- {
		if already[migrations[i].Version] {
			last = &migrations[i]
			break
		}
	}
	if last == nil {
		return "", nil
	}

	tx, err := pool.Begin(ctx)
	if err != nil {
		return "", fmt.Errorf("transaktion fuer rollback von %q starten: %w", last.Version, err)
	}
	if _, err := tx.Exec(ctx, last.DownSQL); err != nil {
		_ = tx.Rollback(ctx)
		return "", fmt.Errorf("migration %q zurueckrollen: %w", last.Version, err)
	}
	if _, err := tx.Exec(ctx, `DELETE FROM `+mg.table+` WHERE version = $1`, last.Version); err != nil {
		_ = tx.Rollback(ctx)
		return "", fmt.Errorf("migration %q aus %s entfernen: %w", last.Version, mg.table, err)
	}
	if err := tx.Commit(ctx); err != nil {
		return "", fmt.Errorf("rollback von %q committen: %w", last.Version, err)
	}
	return last.Version, nil
}