Merge pull request 'common v0.2.0: migrate mit Tabellenname als Parameter' (#385) from feat/common-migrate-v02 into main

.gitea/workflows/common-tag.yml

@@ -22,10 +22,11 @@ jobs:
           go build ./...
           go vet ./...
           go test ./... -count=1
-      - name: Naechsten Patch-Tag setzen
+      - name: Naechsten Tag setzen
         run: |
-          last=$(git tag -l 'common/v0.*' --sort=-v:refname | head -n1)
-          if [ -z "$last" ]; then next="common/v0.1.0"; else
+          minor=$(tr -d '[:space:]' < common/MINOR)
+          last=$(git tag -l "common/v0.${minor}.*" --sort=-v:refname | head -n1)
+          if [ -z "$last" ]; then next="common/v0.${minor}.0"; else
             patch=${last##*.}; base=${last%.*}; next="${base}.$((patch + 1))"
           fi
           git tag "$next"

common/MINOR

@@ -0,0 +1 @@
+2

common/README.md

@@ -22,5 +22,5 @@ Tags `common/v0.X.Y` setzt `.gitea/workflows/common-tag.yml` nach jedem Merge in
 - Nur Code, der in mindestens zwei Modulen byte-gleich (ohne Kommentare) gebraucht wird; fachlich verschiedene Pakete bleiben modul-lokal.
 - Änderungen abwärtskompatibel halten, solange Module nicht gleichzeitig umgestellt werden.
 
-- Jeder PR an `common/` nennt die betroffenen Module (Auswirkungsliste). Breaking Changes nur mit neuer Minor-Version (`v0.2.0`) und Umstellungs-PRs je Modul.
+- Jeder PR an `common/` nennt die betroffenen Module (Auswirkungsliste). Breaking Changes nur mit neuer Minor-Version (`v0.2.0`) und Umstellungs-PRs je Modul; die aktuelle Minor steht in `common/MINOR`, der Tag-Workflow setzt bei einer neuen Minor `v0.<minor>.0`.
 - Sicherheitskritische Pakete (`svcauth`, `tenantregistryclient`, `auditclient`, `policyclient`, `apierror`, `httpharden`) unterliegen dem Pflichttest-Gate (`cmd/pflichttestgate`, Workflow `common-pflichttest-gate.yml`): Änderung ohne begleitende `_test.go` im selben Package schlägt fehl.

common/migrate/migrate.go

@@ -1,6 +1,7 @@
-// Package migrate implements versioned, revertible SQL migrations (no ORM) — a
-// standalone copy of the Core/DMS pattern, since Connect is its own Go module
-// and cannot import other modules' internal/ packages.
+// 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 (
@@ -8,6 +9,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
 	"sort"
 	"strings"
 
@@ -52,47 +54,71 @@ func Load(dir string) ([]Migration, error) {
 	return migrations, nil
 }
 
-// TrackingTable ist modulspezifisch benannt, damit sich Module in
-// derselben Datenbank/demselben Schema nicht gegenseitig den
-// Migrationsfortschritt ueberschreiben (FDN-13).
-const TrackingTable = "connect_schema_migrations"
+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 TrackingTable an. Beim erstmaligen Anlegen wird
+// 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 ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) error {
+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", TrackingTable, err)
+		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('connect_schema_migrations'))`); err != nil {
-		return fmt.Errorf("%s anlegen: %w", TrackingTable, err)
+	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`, TrackingTable).Scan(&existed); err != nil {
-		return fmt.Errorf("%s anlegen: %w", TrackingTable, err)
+	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 `+TrackingTable+` (
+		CREATE TABLE `+mg.table+` (
 			version    TEXT PRIMARY KEY,
 			applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
 		)
 	`); err != nil {
-		return fmt.Errorf("%s anlegen: %w", TrackingTable, err)
+		return fmt.Errorf("%s anlegen: %w", mg.table, err)
 	}
 
 	if len(migrations) > 0 {
@@ -113,7 +139,7 @@ func ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, migrations []M
 			}
 			if hasBase {
 				if _, err := tx.Exec(ctx, `
-					INSERT INTO `+TrackingTable+` (version, applied_at)
+					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)
@@ -124,8 +150,8 @@ func ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, migrations []M
 	return tx.Commit(ctx)
 }
 
-func appliedVersions(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) {
-	rows, err := pool.Query(ctx, `SELECT version FROM `+TrackingTable)
+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)
 	}
@@ -146,11 +172,11 @@ func appliedVersions(ctx context.Context, pool *pgxpool.Pool) (map[string]bool,
 // (Akzeptanzkriterium 2: vorwaerts ausfuehrbar) — bereits angewendete
 // werden uebersprungen, damit Up auf einer leeren UND auf einer bestehenden
 // DB funktioniert (Pruefung 1).
-func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applied []string, err error) {
-	if err := ensureTrackingTable(ctx, pool, migrations); err != nil {
+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 := appliedVersions(ctx, pool)
+	already, err := mg.appliedVersions(ctx, pool)
 	if err != nil {
 		return nil, err
 	}
@@ -167,7 +193,7 @@ func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applie
 			_ = tx.Rollback(ctx)
 			return applied, fmt.Errorf("migration %q anwenden: %w", m.Version, err)
 		}
-		if _, err := tx.Exec(ctx, `INSERT INTO `+TrackingTable+` (version) VALUES ($1)`, m.Version); err != nil {
+		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)
 		}
@@ -182,11 +208,11 @@ func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applie
 // DownOne macht die zuletzt angewendete Migration rueckgaengig
 // (Akzeptanzkriterium 2: rueckwaerts ausfuehrbar) und liefert deren Version,
 // oder "" falls keine Migration angewendet war.
-func DownOne(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (version string, err error) {
-	if err := ensureTrackingTable(ctx, pool, migrations); err != nil {
+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 := appliedVersions(ctx, pool)
+	already, err := mg.appliedVersions(ctx, pool)
 	if err != nil {
 		return "", err
 	}
@@ -210,9 +236,9 @@ func DownOne(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (v
 		_ = tx.Rollback(ctx)
 		return "", fmt.Errorf("migration %q zurueckrollen: %w", last.Version, err)
 	}
-	if _, err := tx.Exec(ctx, `DELETE FROM `+TrackingTable+` WHERE version = $1`, last.Version); err != nil {
+	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, TrackingTable, err)
+		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)

common/migrate/migrate_test.go

@@ -0,0 +1,98 @@
+package migrate
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"gitea.perlbach24.de/scripte/nexarch/common/testdb"
+)
+
+func TestNewValidatesTableName(t *testing.T) {
+	for _, ok := range []string{"dms_schema_migrations", "_x", "a1"} {
+		if _, err := New(ok); err != nil {
+			t.Errorf("New(%q): %v", ok, err)
+		}
+	}
+	for _, bad := range []string{"", "1abc", "Dms", "a-b", "a;drop table x", "a b", "schema.table"} {
+		if _, err := New(bad); err == nil {
+			t.Errorf("New(%q) sollte scheitern", bad)
+		}
+	}
+}
+
+func writeMigrations(t *testing.T, files map[string]string) []Migration {
+	t.Helper()
+	dir := t.TempDir()
+	for name, body := range files {
+		if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	migs, err := Load(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return migs
+}
+
+func TestUpDownAndSeparateTrackingTables(t *testing.T) {
+	ctx := context.Background()
+	pool := testdb.Pool(t, testdb.AdminDSN(t))
+	migs := writeMigrations(t, map[string]string{
+		"0001_a.up.sql":   "CREATE TABLE mig_a (id int)",
+		"0001_a.down.sql": "DROP TABLE mig_a",
+		"0002_b.up.sql":   "CREATE TABLE mig_b (id int)",
+		"0002_b.down.sql": "DROP TABLE mig_b",
+	})
+	one := MustNew("one_schema_migrations")
+	two := MustNew("two_schema_migrations")
+
+	applied, err := one.Up(ctx, pool, migs)
+	if err != nil || len(applied) != 2 {
+		t.Fatalf("Up: applied=%v err=%v", applied, err)
+	}
+	again, err := one.Up(ctx, pool, migs)
+	if err != nil || len(again) != 0 {
+		t.Fatalf("zweites Up soll nichts anwenden: applied=%v err=%v", again, err)
+	}
+
+	var n int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM two_schema_migrations`).Scan(&n); err == nil {
+		t.Fatal("two_schema_migrations darf vor dem ersten Lauf von two nicht existieren")
+	}
+	if _, err := two.DownOne(ctx, pool, migs); err != nil {
+		t.Fatalf("two.DownOne: %v", err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM one_schema_migrations`).Scan(&n); err != nil || n != 2 {
+		t.Fatalf("Fortschritt von one muss unberuehrt sein: n=%d err=%v", n, err)
+	}
+
+	v, err := one.DownOne(ctx, pool, migs)
+	if err != nil || v != "0002_b" {
+		t.Fatalf("DownOne: v=%q err=%v", v, err)
+	}
+}
+
+func TestLegacyTrackingTableIsAdopted(t *testing.T) {
+	ctx := context.Background()
+	pool := testdb.Pool(t, testdb.AdminDSN(t))
+	migs := writeMigrations(t, map[string]string{
+		"0001_a.up.sql":   "CREATE TABLE mig_leg (id int)",
+		"0001_a.down.sql": "DROP TABLE mig_leg",
+	})
+	for _, q := range []string{
+		`CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
+		`INSERT INTO schema_migrations (version) VALUES ('0001_a')`,
+		`CREATE TABLE mig_leg (id int)`,
+	} {
+		if _, err := pool.Exec(ctx, q); err != nil {
+			t.Fatal(err)
+		}
+	}
+	applied, err := MustNew("leg_schema_migrations").Up(ctx, pool, migs)
+	if err != nil || len(applied) != 0 {
+		t.Fatalf("uebernommene Version darf nicht erneut laufen: applied=%v err=%v", applied, err)
+	}
+}